deep-stack/laconicd
3
0 Fork 0
forked from cerc-io/laconicd
Code Pull Requests Activity

Integrate go-nitro

Browse Source 
- Add nitro server and x/nitro module
    - wire go-nitro p2p through cometbft

- Add distsig server, currently WIP
    - integrate DKG and DSS schemes into ABCI methods

- Remove deprecated features
    - crisis module
    - module invariants

- Update to use newer SDK patterns
    - upgrade sdk to v0.53.x
    - custom address codec
    - expand use of depinject
    - migrate e2e tests to system tests
    - use depinject to set up integration tests
    - change reserved protobuf field name `cerc.registry.v1.Record.type`

- Revise & add documentation
    - TransferCoinsToModuleAccount: clarify function

- Update init.sh script
This commit is contained in:
roysc committed 2025-09-21 11:44:44 +08:00
1 parent a8ebc0aceb
commit 5aa2594073
179 files changed
+15631 -3058
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

server/components.go
+120
View File
@@ -0,0 +1,120 @@
package server
import (
"context"
"errors"
"fmt"
"cosmossdk.io/log"
"github.com/cometbft/cometbft/node"
"github.com/cometbft/cometbft/p2p"
"github.com/cosmos/cosmos-sdk/client"
"github.com/cosmos/cosmos-sdk/server"
"github.com/spf13/cobra"
)
// HasContextKey is implemented by servers that have a dedicated key under which they should be
// associated with a context.
type HasContextKey interface {
ContextKey() string
}
type HasP2PReactors interface {
P2PReactors() map[string]p2p.Reactor
}
// maps commands to server components they require
var cmdRequirements = map[*cobra.Command][]ServerComponent{}
// SetRequiredComponents sets required server components for a command
func SetRequiredComponents(cmd *cobra.Command, components ...ServerComponent) {
for _, c := range components {
cmdRequirements[cmd] = append(cmdRequirements[cmd], c)
if startmod, ok := c.(HasStartFlags); ok {
cmd.Flags().AddFlagSet(startmod.StartCmdFlags())
}
}
}
func RequiresComponent(cmd *cobra.Command, c string) bool {
components, ok := cmdRequirements[cmd]
if !ok {
return false
}
for _, mod := range components {
if mod.Name() == c {
return true
}
}
return false
}
type ComponentsCreator func(ConfigMap, client.Context, log.Logger, func(string) bool) ([]ServerComponent, error)
type ServerAux struct {
components []ServerComponent
}
// AddComponents adds hooks to create, start and stop the components returned from the passed
// constructor, and ensures they are available on the command context.
// It returns a node.Option to additionally configure the CometBFT node.
func (s *ServerAux) AddComponents(cmd *cobra.Command, initComponents ComponentsCreator) {
needsComponent := func(name string) bool {
return RequiresComponent(cmd, name)
}
// manage component lifecycle
cmd.PreRunE = func(cmd *cobra.Command, args []string) error {
serverCtx := server.GetServerContextFromCmd(cmd)
clientCtx, err := client.GetClientQueryContext(cmd)
if err != nil {
return err
}
s.components, err = initComponents(
serverCtx.Viper.AllSettings(), clientCtx, serverCtx.Logger, needsComponent,
)
// add available context keys
for _, mod := range s.components {
if ckeymod, ok := mod.(HasContextKey); ok {
cmdCtx := cmd.Context()
if cmdCtx == nil {
cmdCtx = context.Background()
}
cmd.SetContext(context.WithValue(cmdCtx, ckeymod.ContextKey(), mod))
}
}
for _, mod := range s.components {
if err := mod.Start(cmd.Context()); err != nil {
return fmt.Errorf("failed to start server %s: %w", mod.Name(), err)
}
}
return nil
}
startFn := cmd.RunE
cmd.RunE = func(cmd *cobra.Command, args []string) error {
// if we do this in PostRunE, it won't be called if RunE errors
defer func() {
var err error
for _, mod := range s.components {
err = errors.Join(err, mod.Stop(cmd.Context()))
}
if err != nil {
cmd.PrintErrln("failed to stop servers:", err)
}
}()
return startFn(cmd, args)
}
}
// AddReactors adds p2p Reactors for all configured components to a CometBFT node.
// When passed to StartCmdOptions this will be run in start(), after components are initialized in PreRun.
func (s *ServerAux) AddReactors(cmt *node.Node) {
for _, mod := range s.components {
if reactormod, ok := mod.(HasP2PReactors); ok {
opt := node.CustomReactors(reactormod.P2PReactors())
opt(cmt)
}
}
}
server/config.go
+65
View File
@@ -0,0 +1,65 @@
package server
// The utilities in this file are copied from cosmos-sdk server/v2
import (
"fmt"
"github.com/mitchellh/mapstructure"
"github.com/spf13/viper"
)
type ConfigMap map[string]any
// ReadConfig returns a viper instance of the config file
func ReadConfig(configPath string) (*viper.Viper, error) {
v := viper.New()
v.SetConfigType("toml")
v.SetConfigName("config")
v.AddConfigPath(configPath)
if err := v.ReadInConfig(); err != nil {
return nil, fmt.Errorf("failed to read config: %s: %w", configPath, err)
}
v.SetConfigName("app")
if err := v.MergeInConfig(); err != nil {
return nil, fmt.Errorf("failed to merge configuration: %w", err)
}
v.WatchConfig()
return v, nil
}
// UnmarshalSubConfig unmarshals the given (sub) config from the main config (given as a map) into the target.
// If subName is empty, the main config is unmarshaled into the target.
func UnmarshalSubConfig(cfg map[string]any, subName string, target any) error {
var sub any
if subName != "" {
if val, ok := cfg[subName]; ok {
sub = val
}
} else {
sub = cfg
}
// Create a new decoder with custom decoding options
decoder, err := mapstructure.NewDecoder(&mapstructure.DecoderConfig{
DecodeHook: mapstructure.ComposeDecodeHookFunc(
mapstructure.StringToTimeDurationHookFunc(),
mapstructure.StringToSliceHookFunc(","),
),
Result: target,
WeaklyTypedInput: true,
})
if err != nil {
return fmt.Errorf("failed to create decoder: %w", err)
}
// Decode the sub-configuration
if err := decoder.Decode(sub); err != nil {
return fmt.Errorf("failed to decode sub-configuration: %w", err)
}
return nil
}
server/distsig/config.go
+29
View File
@@ -0,0 +1,29 @@
package distsig
import (
"fmt"
"git.vdb.to/cerc-io/laconicd/server"
)
type Config struct {
Enable bool `mapstructure:"enable" toml:"enable" comment:"Enable distributed Schnorr signatures"`
LongtermKey string `mapstructure:"longterm-key" toml:"longterm-key" comment:"The long-term key used in distributed signatures"`
// TODO: set this in genesis
ThresholdRatio float64 `mapstructure:"threshold-ratio" toml:"threshold-ratio" comment:"The ratio of signers required to sign a message"`
}
func DefaultConfig() *Config {
return &Config{
Enable: false,
ThresholdRatio: 4. / 7,
}
}
func UnmarshalConfig(cfg map[string]any) (*Config, error) {
config := DefaultConfig()
if err := server.UnmarshalSubConfig(cfg, componentName, config); err != nil {
return nil, fmt.Errorf("failed to unmarshal %T: %w", config, err)
}
return config, nil
}
server/distsig/distsig_test.go
+175
View File
@@ -0,0 +1,175 @@
package distsig
import (
"errors"
"math/big"
"testing"
"cosmossdk.io/log"
"git.vdb.to/cerc-io/chain-signatures/ethschnorr"
sdktestutil "github.com/cosmos/cosmos-sdk/testutil"
"github.com/stretchr/testify/require"
"git.vdb.to/cerc-io/laconicd/testutil"
)
func NewTestManager(t *testing.T) (*Manager, Point) {
logger := log.NewTestLogger(t)
kr := testutil.NewKeyring()
accounts := sdktestutil.CreateKeyringAccounts(t, kr, 1)
cfg := DefaultConfig()
cfg.LongtermKey = accounts[0].Name
dsm, err := New(logger, cfg, kr)
require.NoError(t, err)
longterm, err := kr.Key(cfg.LongtermKey)
require.NoError(t, err)
longtermPoint, err := KeyRecordToPoint(longterm)
require.NoError(t, err)
return dsm, longtermPoint
}
type round struct {
name string
verify func(*Manager) error
tweak func([]PeerMessages)
}
var (
dkgRounds = []round{
{
name: "deal",
},
{
name: "response",
verify: func(m *Manager) error {
if !m.currentDkg().Certified() {
return errors.New("not certified")
}
return nil
},
},
{
name: "commit",
verify: func(m *Manager) error {
if !m.currentDkg().Finished() {
return errors.New("run not finished")
}
if m.currentDkg().share == nil {
return errors.New("dist key share is absent")
}
return nil
},
},
}
dssRounds = []round{
{
name: "sign",
verify: func(m *Manager) error {
dss := m.sigs[m.completeSigs[0]]
if dss.sig == nil {
return errors.New("signature is nil")
}
dkg, err := m.getDkg(dss.dkgID)
if err != nil {
return err
}
return ethschnorr.Verify(dkg.share.Public(), dss.Message(), dss.sig)
},
},
}
)
func runRound(t *testing.T, r round, members []*Manager) {
buf := make([]PeerMessages, len(members))
for i := range members {
buf[i] = members[i].FlushMessages()
}
if r.tweak != nil {
r.tweak(buf)
}
for i := range members {
for j := range members {
if i == j {
continue
}
require.NoError(t, members[i].ProcessMessages(&buf[j]),
"round=%s, i=%d, j=%d", r.name, i, j)
}
}
if r.verify != nil {
for i, m := range members {
require.NoError(t, r.verify(m), "round=%s, i=%d", r.name, i)
}
}
}
func TestDkgBasic(t *testing.T) {
numMembers := 2
blockHeight := int64(1)
members := make([]*Manager, numMembers)
pubkeys := make([]Point, numMembers)
for i := range members {
members[i], pubkeys[i] = NewTestManager(t)
}
t.Logf("DKG with %d members", numMembers)
for i, m := range members {
require.NoError(t, m.StartDKG(blockHeight, pubkeys), "i=%d", i)
}
for _, r := range dkgRounds {
runRound(t, r, members)
}
require.False(t, members[0].NeedDKG(pubkeys))
require.Error(t, members[0].StartDKG(blockHeight, pubkeys))
// add participants and refresh DKG
for len(members) < 7 {
blockHeight++
m, pub := NewTestManager(t)
members = append(members, m)
pubkeys = append(pubkeys, pub)
t.Logf("DKG with %d members", len(members))
for i, m := range members {
require.True(t, m.NeedDKG(pubkeys), "i=%d", i)
require.NoError(t, m.StartDKG(blockHeight, pubkeys), "i=%d", i)
}
for _, r := range dkgRounds {
runRound(t, r, members)
}
}
}
func TestSignatureBasic(t *testing.T) {
numMembers := 7
members := make([]*Manager, numMembers)
pubkeys := make([]Point, numMembers)
for i := range members {
members[i], pubkeys[i] = NewTestManager(t)
}
for _, m := range members {
require.NoError(t, m.StartDKG(1, pubkeys))
}
for _, r := range dkgRounds {
runRound(t, r, members)
}
msg := big.NewInt(42)
for _, m := range members {
require.NoError(t, m.StartSignature(m.currentDkgID, msg))
}
runRound(t, dssRounds[0], members)
require.Error(t, members[0].StartSignature(0, msg))
require.NoError(t, members[0].StartDKG(2, pubkeys[:3]))
require.Error(t, members[0].StartSignature(2, msg))
}
server/distsig/flags.go
+19
View File
@@ -0,0 +1,19 @@
package distsig
import (
"fmt"
"github.com/spf13/pflag"
)
func prefix(f string) string {
return fmt.Sprintf("%s.%s", componentName, f)
}
var (
FlagLongtermKey = prefix("longterm-key")
)
func AddNitroFlags(flags *pflag.FlagSet) {
flags.String(FlagLongtermKey, "", "The long-term key used in distributed signatures")
}
server/distsig/keygen.go
+224
View File
@@ -0,0 +1,224 @@
package distsig
import (
"fmt"
dkg "go.dedis.ch/kyber/v3/share/dkg/rabin"
)
// DkgMessages collects local DKG messages that are pending broadcast.
type DkgMessages struct {
runID DkgRunID
messages []dkgMessage
}
// DKG steps (actions) that are executed by the DKG protocol
type dkgStep int
const (
dkg_deal dkgStep = iota
dkg_certify
dkg_commit
dkg_finish
dkg_done
)
type dkgRun struct {
*dkg.DistKeyGenerator
// runID RunID
msgBuffer []dkgMessage
step dkgStep
share *dkg.DistKeyShare
}
type dkgMessage interface {
send(*dkgRun) (dkgMessage, error)
from() uint32
target() uint32
}
// prepareMessages produces DKG state transitions that don't rely on peer inputs, i.e. the deal or
// commit steps. This is called during ExtendVote by all validators.
func (d *dkgRun) prepareMessages() error {
out, err := d.handleMessage(nil)
if err != nil {
return err
}
if out != nil {
d.msgBuffer = append(d.msgBuffer, out...)
}
return nil
}
func (d *dkgRun) flushMessages() []dkgMessage {
buf := d.msgBuffer
d.msgBuffer = nil
return buf
}
// processMessages processes incoming DKG messages and produces outgoing messages. This is called
// during VerifyVoteExtension by all validators (TODO: verify).
func (d *dkgRun) processMessages(in []dkgMessage) error {
for _, msg := range in {
out, err := d.handleMessage(msg)
if err != nil {
return err
}
if out != nil {
d.msgBuffer = append(d.msgBuffer, out...)
}
}
return nil
}
func (d *dkgRun) handleMessage(in dkgMessage) ([]dkgMessage, error) {
// fmt.Printf("[%d, %7s] handle: %T", d.index, d.step, in)
// if in != nil {
// fmt.Printf(" (from %d to %+v)", in.from(), in.target())
// }
// fmt.Println()
// nil input means we try to execute a step requiring no peer input
if in == nil {
var out []dkgMessage
switch d.step {
case dkg_deal:
deals, err := d.Deals()
if err != nil {
return nil, err
}
for index, deal := range deals {
out = append(out, msgDeal{index: uint32(index), msg: deal})
}
case dkg_commit:
sc, err := d.SecretCommits()
if err != nil {
return nil, err
}
out = []dkgMessage{msgSecretCommits{msg: sc}}
case dkg_done:
dks, err := d.DistKeyShare()
if err != nil {
return nil, err
}
d.share = dks
default:
return nil, fmt.Errorf("unexpected stage: %v", d.step)
}
d.step = nextStep(d.step)
return out, nil
}
var out []dkgMessage
if msg, err := in.send(d); err != nil {
return nil, err
} else if msg != nil {
out = []dkgMessage{msg}
}
if (d.step == dkg_certify && d.Certified()) ||
(d.step == dkg_finish && d.Finished()) {
d.step = nextStep(d.step)
msgs, err := d.handleMessage(nil)
if err != nil {
return nil, err
}
out = append(out, msgs...)
}
return out, nil
}
func nextStep(step dkgStep) dkgStep {
if step == dkg_done {
return dkg_done
}
return step + 1
}
type (
msgDeal struct {
msg *dkg.Deal
index uint32 // recipient index
}
msgResponse struct{ msg *dkg.Response }
msgJustification struct{ msg *dkg.Justification }
msgSecretCommits struct{ msg *dkg.SecretCommits }
msgComplaintCommits struct{ msg *dkg.ComplaintCommits }
msgReconstructCommits struct{ msg *dkg.ReconstructCommits }
)
func (m msgDeal) send(d *dkgRun) (dkgMessage, error) {
if m.index != uint32(d.Index()) {
return nil, nil
}
out, err := d.ProcessDeal(m.msg)
if err != nil {
return nil, err
}
return msgResponse{out}, nil
}
func (m msgResponse) send(d *dkgRun) (dkgMessage, error) {
if out, err := d.ProcessResponse(m.msg); err != nil {
return nil, err
} else if out != nil {
return msgJustification{out}, nil
}
return nil, nil
}
func (m msgJustification) send(d *dkgRun) (dkgMessage, error) {
return nil, d.ProcessJustification(m.msg)
}
func (m msgSecretCommits) send(d *dkgRun) (dkgMessage, error) {
if out, err := d.ProcessSecretCommits(m.msg); err != nil {
return nil, err
} else if out != nil {
return msgComplaintCommits{out}, nil
}
return nil, nil
}
func (m msgComplaintCommits) send(d *dkgRun) (dkgMessage, error) {
out, err := d.ProcessComplaintCommits(m.msg)
if err != nil {
return nil, err
}
return msgReconstructCommits{out}, nil
}
func (m msgReconstructCommits) send(d *dkgRun) (dkgMessage, error) {
return nil, d.ProcessReconstructCommits(m.msg)
}
func (d dkgStep) String() string {
switch d {
case dkg_deal:
return "deal"
case dkg_certify:
return "certify"
case dkg_commit:
return "commit"
case dkg_finish:
return "finish"
case dkg_done:
return "done"
default:
return "unknown"
}
}
// for debugging
func (m msgDeal) from() uint32 { return m.msg.Index }
func (m msgResponse) from() uint32 { return m.msg.Response.Index }
func (m msgJustification) from() uint32 { return m.msg.Justification.Index }
func (m msgSecretCommits) from() uint32 { return m.msg.Index }
func (m msgComplaintCommits) from() uint32 { return m.msg.Index }
func (m msgReconstructCommits) from() uint32 { return m.msg.Index }
func (m msgDeal) target() uint32 { return m.index }
func (m msgResponse) target() uint32 { return m.msg.Index }
func (m msgJustification) target() uint32 { return m.msg.Index }
func (m msgSecretCommits) target() uint32 { return 0 }
func (m msgComplaintCommits) target() uint32 { return m.msg.DealerIndex }
func (m msgReconstructCommits) target() uint32 { return m.msg.DealerIndex }
server/distsig/manager.go
+268
View File
@@ -0,0 +1,268 @@
package distsig
import (
"context"
"fmt"
"math"
"math/big"
cmtcrypto "github.com/cometbft/cometbft/crypto"
"github.com/ethereum/go-ethereum/common"
dkg "go.dedis.ch/kyber/v3/share/dkg/rabin"
"cosmossdk.io/log"
"github.com/cosmos/cosmos-sdk/crypto/keyring"
clientdss "git.vdb.to/cerc-io/chain-signatures/ethdss"
"git.vdb.to/cerc-io/chain-signatures/ethschnorr"
"git.vdb.to/cerc-io/laconicd/server"
"git.vdb.to/cerc-io/laconicd/utils"
)
type (
DealMap = map[int]*dkg.Deal
DkgRunID int64
SigRunID string
)
const (
componentName = "distsig"
)
type Manager struct {
config *Config
logger log.Logger
keyring keyring.Keyring
longtermKey Scalar // only initialized after Start()
longtermPubKey cmtcrypto.PubKey
dkgs map[DkgRunID]*dkgRun
currentDkgID DkgRunID // zero indicates no current run
sigs map[SigRunID]*sigRun
completeSigs []SigRunID
}
type PeerMessages struct {
dkg *DkgMessages
dss []SigMessages
}
func New(logger log.Logger, globalConfig server.ConfigMap, kr keyring.Keyring) (*Manager, error) {
config, err := UnmarshalConfig(globalConfig)
if err != nil {
return nil, err
}
m := &Manager{
config: config,
keyring: kr,
dkgs: make(map[DkgRunID]*dkgRun),
sigs: make(map[SigRunID]*sigRun),
}
m.logger = logger.With(log.ModuleKey, m.Name())
return m, nil
}
func (*Manager) Name() string { return componentName }
func (m *Manager) Start(ctx context.Context) error {
if !m.config.Enable {
m.logger.Info(fmt.Sprintf("%s server is disabled via config", m.Name()))
return nil
}
if m.config.LongtermKey == "" {
return fmt.Errorf("missing longterm key")
}
longtermPrivKey, err := utils.ExtractPrivateKeyByUid(m.keyring, m.config.LongtermKey)
if err != nil {
return fmt.Errorf("failed to extract longterm key: %w", err)
}
m.longtermKey = suite.Scalar().SetBytes(longtermPrivKey.Bytes())
m.longtermPubKey = longtermPrivKey.PubKey()
return nil
}
func (*Manager) Stop(context.Context) error { return nil }
func (m *Manager) Config() any {
if m.config == nil {
return DefaultConfig()
}
return m.config
}
func (m *Manager) getDkg(runid DkgRunID) (*dkgRun, error) {
if runid == 0 {
// runid = m.currentDkg
return nil, fmt.Errorf("invalid run ID")
}
if run, ok := m.dkgs[runid]; ok {
return run, nil
}
return nil, fmt.Errorf("DKG not initialized for run: %v", runid)
}
func (m *Manager) currentDkg() *dkgRun {
run, err := m.getDkg(m.currentDkgID)
if err != nil {
panic(err)
}
return run
}
func (m *Manager) LongtermPublicKey() cmtcrypto.PubKey {
return m.longtermPubKey
}
func (m *Manager) LongtermEthAddress() common.Address {
pubkey, err := SuitePublicKeyFromBytes(m.LongtermPublicKey().Bytes())
if err != nil {
panic(fmt.Errorf("failed to parse longterm pubkey: %w", err))
}
return pubkey.Address()
}
func thresholdForRatio(m int, tr float64) int {
return int(math.Ceil(float64(m) * tr))
}
func (m *Manager) initDKG(members []Point) (*dkgRun, error) {
t := thresholdForRatio(len(members), m.config.ThresholdRatio)
keygen, err := dkg.NewDistKeyGenerator(suite.(dkg.Suite), m.longtermKey, members, t)
if err != nil {
return nil, err
}
return &dkgRun{DistKeyGenerator: keygen}, nil
}
// NeedDKG returns true if DKG needs to be run for the given participants.
func (m *Manager) NeedDKG(pubkeys []Point) bool {
return m.currentDkgID == 0 || !equalPoints(m.currentDkg().Participants(), pubkeys)
}
// StartDKG begins a new DKG run including the given participants.
func (m *Manager) StartDKG(block int64, pubkeys []Point) error {
// TODO: generate runid from block height and hash of pubkeys?
runid := DkgRunID(block)
if _, ok := m.dkgs[runid]; ok {
return fmt.Errorf("DKG run %v already exists", runid)
}
if len(pubkeys) < 2 {
m.currentDkgID = 0
m.logger.Debug("Too few participants for distributed signature")
return nil
}
run, err := m.initDKG(pubkeys)
if err != nil {
return err
}
m.dkgs[runid] = run
m.currentDkgID = runid
return run.prepareMessages()
}
func (m *Manager) initDSS(dkgid DkgRunID, msg *big.Int) (*sigRun, error) {
run, err := m.getDkg(dkgid)
if err != nil {
return nil, err
}
if !run.Finished() {
return nil, fmt.Errorf("DKG run has not finished: %v", dkgid)
}
random, err := run.DistKeyShare()
if err != nil {
return nil, err
}
dss, err := clientdss.NewDSS(clientdss.DSSArgs{
Secret: m.longtermKey,
Participants: run.Participants(),
Long: run.share,
Random: random,
Msg: msg,
T: run.Threshold(),
// Qualified: run.QUAL(),
})
if err != nil {
return nil, err
}
return &sigRun{DSS: dss, dkgID: dkgid}, nil
}
// StartSignature begins a DSS run for the given message and DKG state.
func (m *Manager) StartSignature(msg *big.Int) error {
if m.currentDkgID == 0 {
return fmt.Errorf("no distributed key prepared")
}
run, err := m.initDSS(m.currentDkgID, msg)
if err != nil {
return err
}
sigid := run.SigRunID()
m.sigs[sigid] = run
return run.prepareMessages()
}
func (m *Manager) CompletedSignatures() map[SigRunID]ethschnorr.Signature {
var ret map[SigRunID]ethschnorr.Signature
for _, id := range m.completeSigs {
run, ok := m.sigs[id]
if !ok {
panic(fmt.Errorf("DSS run not found: %v", id))
}
if run.sig == nil {
panic(fmt.Errorf("DSS signature not completed: %v", id))
}
// Deterministically pick a participant responsible for submitting the signature
session := new(big.Int).SetBytes(run.SessionID()[:8])
submitterIdx := session.Uint64() % uint64(len(run.Participants()))
if run.Index() == int(submitterIdx) {
ret[id] = run.sig
}
}
return ret
}
// FlushMessages flushes the message buffers for any active DKG and DSS runs.
func (m *Manager) FlushMessages() PeerMessages {
var ret PeerMessages
if buf := m.currentDkg().flushMessages(); len(buf) != 0 {
ret.dkg = &DkgMessages{runID: m.currentDkgID, messages: buf}
}
for id, run := range m.sigs {
if buf := run.flushMessages(); buf != nil {
ret.dss = append(ret.dss, SigMessages{id, buf})
}
}
return ret
}
// ProcessMessages processes DKG and DSS peer messages.
func (m *Manager) ProcessMessages(dm *PeerMessages) error {
if dkg := dm.dkg; dkg != nil {
run, err := m.getDkg(dkg.runID)
if err != nil {
return err
}
if err := run.processMessages(dkg.messages); err != nil {
return err
}
}
for _, dss := range dm.dss {
run, ok := m.sigs[dss.runID]
if !ok {
return fmt.Errorf("DSS not initialized for run: %v", dss.runID)
}
if err := run.processMessage(dss.messages); err != nil {
return err
}
if run.sig != nil {
m.completeSigs = append(m.completeSigs, dss.runID)
}
}
return nil
}
func (pm PeerMessages) Empty() bool {
return pm.dkg == nil && len(pm.dss) == 0
}
server/distsig/signature.go
+61
View File
@@ -0,0 +1,61 @@
package distsig
import (
"fmt"
clientdss "git.vdb.to/cerc-io/chain-signatures/ethdss"
"git.vdb.to/cerc-io/chain-signatures/ethschnorr"
)
type SigMessages struct {
runID SigRunID
messages sigMessage
}
type sigStep int
const (
sig_partial sigStep = iota
sig_complete
)
type sigRun struct {
*clientdss.DSS
msgBuffer sigMessage
dkgID DkgRunID
step sigStep
sig ethschnorr.Signature
}
type sigMessage = *clientdss.PartialSig
func (d *sigRun) SigRunID() SigRunID {
return SigRunID(fmt.Sprintf("%020d-%x", d.dkgID, d.SessionID()[:8]))
}
func (d *sigRun) prepareMessages() error {
var err error
d.msgBuffer, err = d.PartialSig()
return err
}
func (d *sigRun) flushMessages() sigMessage {
buf := d.msgBuffer
d.msgBuffer = nil
return buf
}
func (d *sigRun) processMessage(in sigMessage) error {
err := d.ProcessPartialSig(in)
if err != nil {
return err
}
// return d.EnoughPartialSig(), nil
if d.EnoughPartialSig() {
d.sig, err = d.Signature()
return err
}
return nil
}
server/distsig/suite.go
+39
View File
@@ -0,0 +1,39 @@
package distsig
import (
"fmt"
"git.vdb.to/cerc-io/chain-signatures/secp256k1"
"github.com/cosmos/cosmos-sdk/crypto/keyring"
"go.dedis.ch/kyber/v3"
"go.dedis.ch/kyber/v3/suites"
)
type (
Scalar = kyber.Scalar
Point = kyber.Point
PublicKey = secp256k1.PublicKey
)
var (
suite suites.Suite = secp256k1.NewBlakeKeccackSecp256k1()
NewScalar = suite.Scalar
NewPoint = suite.Point
// Note: the compressed encoding of pubkeys used by Cosmos SDK and our library (based on chainlink)
// are the same. See gitlab.com/yawning/secp256k1-voi/secec
SuitePublicKeyFromBytes = secp256k1.NewPublicKeyFromBytes
)
func KeyRecordToPoint(longterm *keyring.Record) (kyber.Point, error) {
pubkey, err := longterm.GetPubKey()
if err != nil {
return nil, fmt.Errorf("failed to access public key: %w", err)
}
suitePubkey, err := SuitePublicKeyFromBytes(pubkey.Bytes())
if err != nil {
return nil, fmt.Errorf("failed to decode public key: %w", err)
}
return suitePubkey.Point()
}
server/distsig/utils.go
+13
View File
@@ -0,0 +1,13 @@
package distsig
func equalPoints(a, b []Point) bool {
if len(a) != len(b) {
return false
}
for i, p := range a {
if !p.Equal(b[i]) {
return false
}
}
return true
}
server/gas.go
+36
View File
@@ -0,0 +1,36 @@
package server
import (
"context"
"cosmossdk.io/core/gas"
sdk "github.com/cosmos/cosmos-sdk/types"
)
var _ gas.Service = gasService{}
// Placeholder gas service, in anticipation of future use
// https://github.com/cosmos/cosmos-sdk/pull/16310
type gasService struct{}
func NewGasService() gas.Service { return gasService{} }
func (gasService) GetGasMeter(ctx context.Context) gas.Meter {
c := sdk.UnwrapSDKContext(ctx)
return c.GasMeter()
}
func (gasService) WithGasMeter(ctx context.Context, meter gas.Meter) context.Context {
c := sdk.UnwrapSDKContext(ctx)
return c.WithGasMeter(meter)
}
// deprecated https://github.com/cosmos/cosmos-sdk/issues/19793
func (gasService) GetBlockGasMeter(context.Context) gas.Meter {
return nil
}
// deprecated https://github.com/cosmos/cosmos-sdk/issues/19793
func (gasService) WithBlockGasMeter(ctx context.Context, meter gas.Meter) context.Context {
return ctx
}
server/nitro/command_utils.go
+230
View File
@@ -0,0 +1,230 @@
package nitro
import (
"fmt"
"math/big"
"runtime"
"strings"
"time"
"github.com/cometbft/cometbft/p2p"
"github.com/cosmos/cosmos-sdk/client"
"github.com/cosmos/cosmos-sdk/client/tx"
sdk "github.com/cosmos/cosmos-sdk/types"
"github.com/ethereum/go-ethereum/common"
"github.com/spf13/cobra"
"github.com/statechannels/go-nitro/protocols"
nitrotypes "github.com/statechannels/go-nitro/types"
"git.vdb.to/cerc-io/laconicd/server/relay"
"git.vdb.to/cerc-io/laconicd/utils"
)
const txTimeoutDuration = 5 * time.Second
func GetServerFromCmd(cmd *cobra.Command) (*Server, error) {
s, err := utils.GetFromContext[Server](cmd.Context(), ServerContextKey)
if err != nil {
return nil, err
}
for !s.Ready() {
runtime.Gosched()
}
return s, nil
}
func delayAfterUpdate() {
// HACK: delay so remote server status is synced before we return
// TODO: correct solution is passing messages with consistency guarantees,
// i.e. using txs to pass nitro messages and integrating state into chain storage
time.Sleep(500 * time.Millisecond)
}
// waitForObjective waits for an objective to complete or fail using select pattern
func waitForObjective[Info any](
failedChan <-chan protocols.ObjectiveId,
objective protocols.ObjectiveId,
updateChan <-chan Info,
statusCheck func(Info) bool,
) error {
for {
select {
case update := <-updateChan:
if statusCheck(update) {
delayAfterUpdate()
return nil
}
case failedObjective := <-failedChan:
if failedObjective == objective {
return fmt.Errorf("objective failed for unspecified reason: %s", objective)
}
}
}
}
func submitNitroTx(
clientCtx client.Context,
cmd *cobra.Command,
msg MsgWrapper,
unordered bool,
) error {
if msg.SignerField != nil {
ac := utils.NewAddressCodec()
var err error
*msg.SignerField, err = ac.BytesToString(clientCtx.FromAddress)
if err != nil {
return err
}
}
factory, err := tx.NewFactoryCLI(clientCtx, cmd.Flags())
if err != nil {
return err
}
if unordered {
deadline := time.Now().Add(txTimeoutDuration)
factory = factory.WithUnordered(true).WithTimeoutTimestamp(deadline)
}
if err := tx.GenerateOrBroadcastTxWithFactory(clientCtx, factory, msg.Msg); err != nil {
return err
}
return nil
}
// setupNitroCommand sets up common infrastructure for nitro commands
func setupNitroCommand(cmd *cobra.Command) (*client.Context, *Server, func() error, error) {
clientCtx, err := client.GetClientTxContext(cmd)
if err != nil {
return nil, nil, nil, err
}
s, err := GetServerFromCmd(cmd)
if err != nil {
return nil, nil, nil, err
}
relayServer, err := relay.GetServerFromCmd(cmd)
if err != nil {
return nil, nil, nil, err
}
sw, err := connectAsPeer(clientCtx, relayServer, s.P2PReactors())
if err != nil {
return nil, nil, nil, err
}
return &clientCtx, s, sw.Stop, nil
}
// setupNitroQueryCommand sets up infrastructure for nitro query commands (no tx context needed)
func setupNitroQueryCommand(cmd *cobra.Command, needp2p bool) (*Server, func() error, error) {
s, err := GetServerFromCmd(cmd)
if err != nil {
return nil, nil, err
}
clientCtx, err := client.GetClientQueryContext(cmd)
if err != nil {
return nil, nil, err
}
if !needp2p {
return s, func() error { return nil }, nil
}
relayServer, err := relay.GetServerFromCmd(cmd)
if err != nil {
return nil, nil, err
}
sw, err := connectAsPeer(clientCtx, relayServer, s.P2PReactors())
if err != nil {
return nil, nil, err
}
return s, sw.Stop, nil
}
// resolveTokens resolves amount and asset address to Nitro funds
func resolveTokens(s *Server, amountStr, assetAddr string) (nitrotypes.Funds, error) {
var amount *big.Int
var asset common.Address
if assetAddr != "" {
// Parse asset address
if !common.IsHexAddress(assetAddr) {
return nitrotypes.Funds{}, fmt.Errorf("invalid asset address: %s", assetAddr)
}
asset = common.HexToAddress(assetAddr)
// A token specified by address must have an integer amount
amount = new(big.Int)
if _, ok := amount.SetString(amountStr, 10); !ok {
return nitrotypes.Funds{}, fmt.Errorf("invalid amount for token %s: %s", assetAddr, amountStr)
}
if amount.Sign() <= 0 {
return nitrotypes.Funds{}, fmt.Errorf("amount must be positive for token %s: %s", assetAddr, amountStr)
}
} else {
// Parse as (known) coins and convert
coins, err := sdk.ParseCoinsNormalized(amountStr)
if err != nil {
return nitrotypes.Funds{}, fmt.Errorf("failed to parse coins: %w", err)
}
if len(coins) != 1 {
return nitrotypes.Funds{}, fmt.Errorf("only one asset is supported")
}
// Convert denomination to token address
asset, err = denomToToken(coins[0].Denom)
if err != nil {
return nitrotypes.Funds{}, err
}
amount = coins[0].Amount.BigInt()
}
return nitrotypes.Funds{asset: amount}, nil
}
// map denoms to token addresses
// TODO nitro should control this mapping, and resolution should happen in the module logic
func denomToToken(denom string) (common.Address, error) {
_tokens := map[string]common.Address{
"eth": {},
}
token, ok := _tokens[strings.ToLower(denom)]
if !ok {
return common.Address{}, fmt.Errorf("unknown token %s", denom)
}
return token, nil
}
// connects as a peer to the relay server, initializing the P2P switch
func connectAsPeer(
clientCtx client.Context, relayServer *relay.Server, reactors map[string]p2p.Reactor,
) (*relay.Switch, error) {
// initialize the switch using the target's node info
status, err := clientCtx.Client.Status(clientCtx.CmdContext)
if err != nil {
return nil, err
}
nodeInfo := status.NodeInfo
sw, err := relayServer.GetSwitch(clientCtx.ChainID, nodeInfo.ProtocolVersion.Block)
if err != nil {
return nil, fmt.Errorf("error creating switch: %w", err)
}
for name, reactor := range reactors {
if err := sw.AddReactor(name, reactor); err != nil {
return nil, fmt.Errorf("error adding reactor: %w", err)
}
}
if err := sw.Start(); err != nil {
return nil, fmt.Errorf("error starting switch: %w", err)
}
peerAddr := p2p.IDAddressString(nodeInfo.ID(), nodeInfo.ListenAddr)
// s.logger.Debug("dialing peer", "peer", peerAddr)
if err := sw.Dial(peerAddr); err != nil {
return nil, fmt.Errorf("error dialing peer: %w", err)
}
return sw, nil
}
server/nitro/config.go
+80
View File
@@ -0,0 +1,80 @@
package nitro
import (
"errors"
"fmt"
"git.vdb.to/cerc-io/laconicd/server"
)
type Config struct {
Enable bool `mapstructure:"enable" toml:"enable" comment:"Enable Nitro state channel functionality"`
EthKey string `mapstructure:"eth-key" toml:"eth-key" comment:"The private key used when interacting with the Ethereum chain and for our identity as a participant in the Nitro protocol."`
// UseDistsig bool `mapstructure:"use-distsig" toml:"use-distsig" comment:"Whether to use distributed signatures to authenticate Nitro actions."`
// EthChainID string
EthURL string `mapstructure:"eth-url" toml:"eth-url" comment:"The URL of the Ethereum node to connect to."`
EthAuthToken string `mapstructure:"eth-auth-token" toml:"eth-auth-token" comment:"The bearer token used for auth in requests to the Ethereum chain's RPC endpoint."`
EthStartBlock uint64 `mapstructure:"eth-start-block" toml:"eth-start-block" comment:"Ethereum block number to start listening for Nitro Adjudicator events."`
// TODO: move to module params?
EthNaAddress string `mapstructure:"eth-na-address" toml:"eth-na-address" comment:"Ethereum address of the Nitro Adjudicator contract."`
EthVpaAddress string `mapstructure:"eth-vpa-address" toml:"eth-vpa-address" comment:"Ethereum address of the Virtual Payment App contract."`
EthCaAddress string `mapstructure:"eth-ca-address" toml:"eth-ca-address" comment:"Ethereum address of the Consensus App contract."`
// P2P Rate Limiting
P2PRateLimitEnable bool `mapstructure:"p2p-rate-limit-enable" toml:"p2p-rate-limit-enable" comment:"Enable send-side rate limiting for P2P messages."`
P2PRateLimitRate float64 `mapstructure:"p2p-rate-limit-rate" toml:"p2p-rate-limit-rate" comment:"Maximum number of P2P messages per second (e.g., 10.0 for 10 messages/second)."`
P2PRateLimitBurst int `mapstructure:"p2p-rate-limit-burst" toml:"p2p-rate-limit-burst" comment:"Maximum burst size for P2P rate limiter."`
}
func DefaultConfig() *Config {
return &Config{
Enable: true,
// UseDistsig: false,
// EthURL: "ws://127.0.0.1:8545",
// Default rate limiting: 10 messages/second with burst of 20
P2PRateLimitEnable: true,
P2PRateLimitRate: 10.0,
P2PRateLimitBurst: 20,
}
}
func (c Config) Validate() error {
if !c.Enable {
return nil
}
if c.EthKey == "" {
return errors.New("nitro.eth-key must be set")
}
// TODO With distsig enabled, eth-key will be a mutually exclusive setting, and authentication
// for signing groups will be done using longterm key
//
// if c.UseDistsig && c.EthKey != "" {
// return errors.New("nitro.eth-key should not be set when distsig is enabled")
// }
// Validate rate limiting config
if c.P2PRateLimitEnable {
if c.P2PRateLimitRate <= 0 {
return errors.New("nitro.p2p-rate-limit-rate must be positive when rate limiting is enabled")
}
if c.P2PRateLimitBurst <= 0 {
return errors.New("nitro.p2p-rate-limit-burst must be positive when rate limiting is enabled")
}
}
return nil
}
func UnmarshalConfig(cfg map[string]any) (*Config, error) {
config := DefaultConfig()
if len(cfg) > 0 {
if err := server.UnmarshalSubConfig(cfg, serverName, config); err != nil {
return nil, fmt.Errorf("failed to unmarshal %T: %w", config, err)
}
}
return config, nil
}
server/nitro/config_test.go
+71
View File
@@ -0,0 +1,71 @@
package nitro
import (
"testing"
"github.com/stretchr/testify/assert"
)
func TestConfigValidation(t *testing.T) {
tests := []struct {
name string
config Config
expectErr bool
}{
{
name: "valid config with rate limiting",
config: Config{
Enable: true,
EthKey: "test-eth-key",
P2PRateLimitEnable: true,
P2PRateLimitRate: 10.0,
P2PRateLimitBurst: 20,
},
expectErr: false,
},
{
name: "invalid rate - zero",
config: Config{
Enable: true,
EthKey: "test-eth-key",
P2PRateLimitEnable: true,
P2PRateLimitRate: 0,
P2PRateLimitBurst: 20,
},
expectErr: true,
},
{
name: "invalid burst - zero",
config: Config{
Enable: true,
EthKey: "test-eth-key",
P2PRateLimitEnable: true,
P2PRateLimitRate: 10.0,
P2PRateLimitBurst: 0,
},
expectErr: true,
},
{
name: "rate limiting disabled - no validation needed",
config: Config{
Enable: true,
EthKey: "test-eth-key",
P2PRateLimitEnable: false,
P2PRateLimitRate: 0, // Invalid but should be ignored
P2PRateLimitBurst: 0, // Invalid but should be ignored
},
expectErr: false,
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
err := tt.config.Validate()
if tt.expectErr {
assert.Error(t, err)
} else {
assert.NoError(t, err)
}
})
}
}
server/nitro/flags.go
+30
View File
@@ -0,0 +1,30 @@
package nitro
import (
"fmt"
"github.com/spf13/pflag"
)
// start flags are prefixed with the server name
func prefix(f string) string {
return fmt.Sprintf("%s.%s", serverName, f)
}
var (
FlagEthKey = prefix("eth-key")
FlagEthURL = prefix("eth-url")
FlagEthStartBlock = prefix("eth-start-block")
FlagEthAuthToken = prefix("eth-auth-token")
// FlagEthNaAddress = prefix("eth-na-address")
// FlagEthVpaAddress = prefix("eth-vpa-address")
// FlagEthCaAddress = prefix("eth-ca-address")
)
func AddFlags(f *pflag.FlagSet) {
f.String(FlagEthKey, "", "name of private key to use when interacting with the Ethereum chain")
f.String(FlagEthURL, "ws://127.0.0.1:8545", "URL of the Ethereum node to connect to")
f.String(FlagEthAuthToken, "", "bearer token used for auth in requests to the Ethereum chain's RPC endpoint")
f.Uint64(FlagEthStartBlock, 0, "Ethereum block number to start listening for Nitro Adjudicator events")
}
server/nitro/p2p.go
+286
View File
@@ -0,0 +1,286 @@
package nitro
import (
"context"
"errors"
"fmt"
"sync"
"time"
"cosmossdk.io/log"
"github.com/cometbft/cometbft/p2p"
"github.com/cosmos/gogoproto/proto"
nitrop2p "github.com/statechannels/go-nitro/node/engine/messageservice"
"github.com/statechannels/go-nitro/protocols"
nitrotypes "github.com/statechannels/go-nitro/types"
"golang.org/x/time/rate"
)
const (
P2PMessageChannel = byte(0x80)
// SigRequestChannel = byte(0x81)
maxMsgSize = 1024 * 1024
msgBufSize = 1000
limiterTimeout = 5 * time.Second
)
var (
_ p2p.Reactor = (*p2pReactor)(nil)
_ nitrop2p.MessageService = (*p2pMsgService)(nil)
)
// p2pReactor handles Nitro P2P messages via CometBFT
type p2pReactor struct {
p2p.BaseReactor
incoming chan nitrop2p.Message
outgoing map[p2p.ID]chan nitrop2p.Message
sendQueue chan nitrop2p.Message
rateLimiter *rate.Limiter
config *Config
logger log.Logger
mtx sync.RWMutex
wg sync.WaitGroup
}
type p2pMsgService struct {
id string
logger log.Logger
reactor *p2pReactor // one reactor per msg service
}
type streamDesc struct {
id byte
msgType proto.Message
}
func (sd streamDesc) StreamID() byte { return sd.id }
// TODO: convert Message to protobuf
func (sd streamDesc) MessageType() proto.Message { return sd.msgType }
// Reactor
func newReactor(config *Config, logger log.Logger) *p2pReactor {
var rateLimiter *rate.Limiter
if config.P2PRateLimitEnable {
rateLimiter = rate.NewLimiter(rate.Limit(config.P2PRateLimitRate), config.P2PRateLimitBurst)
}
ret := &p2pReactor{
incoming: make(chan nitrop2p.Message, msgBufSize),
outgoing: make(map[p2p.ID]chan nitrop2p.Message),
sendQueue: make(chan nitrop2p.Message, msgBufSize),
rateLimiter: rateLimiter,
config: config,
logger: logger,
}
ret.BaseReactor = *p2p.NewBaseReactor("nitro-reactor", ret)
return ret
}
// StreamDescriptors returns the stream descriptor for Nitro messages.
func (r *p2pReactor) GetChannels() []*p2p.ChannelDescriptor {
return []*p2p.ChannelDescriptor{
{
ID: P2PMessageChannel,
Priority: 5,
SendQueueCapacity: 100,
RecvBufferCapacity: 1000,
RecvMessageCapacity: maxMsgSize,
MessageType: &P2PMessage{},
},
// TODO Nitro peer info channel
}
}
// AddPeer begins sending messages to a peer.
func (r *p2pReactor) AddPeer(peer p2p.Peer) {
if !r.IsRunning() {
return
}
peerCh := make(chan nitrop2p.Message, msgBufSize)
r.mtx.Lock()
r.outgoing[peer.ID()] = peerCh
r.mtx.Unlock()
// Start goroutine to handle sending to this specific peer
go r.handlePeer(peer, peerCh)
}
func (r *p2pReactor) handlePeer(peer p2p.Peer, peerCh chan nitrop2p.Message) {
logger := r.logger.With("peer", peer)
defer func() {
r.mtx.Lock()
defer r.mtx.Unlock()
if _, ok := r.outgoing[peer.ID()]; ok {
close(peerCh)
delete(r.outgoing, peer.ID())
}
}()
for {
if !peer.IsRunning() || !r.IsRunning() {
return
}
inner_loop:
for {
select {
case msg := <-peerCh:
encoded, err := msg.Serialize()
if err != nil {
logger.Error("Failed to serialize message", "msg", msg, "err", err)
continue inner_loop
}
sent := peer.Send(p2p.Envelope{
Message: &P2PMessage{Content: encoded},
ChannelID: P2PMessageChannel,
})
if !sent {
logger.Error("Failed to send message", "msg", msg)
return
}
default:
break inner_loop
}
}
}
}
// Receive handles an envelope received from any connected peer on any registered channel.
func (r *p2pReactor) Receive(e p2p.Envelope) {
if !r.IsRunning() {
r.Logger.Debug("Receive", "src", e.Src, "chId", e.ChannelID)
return
}
switch e.ChannelID {
case P2PMessageChannel:
switch msg := e.Message.(type) {
case *P2PMessage:
nitroMsg, err := protocols.DeserializeMessage(msg.Content)
if err != nil {
r.Logger.Error("Failed to deserialize message", "err", err)
r.Switch.StopPeerForError(e.Src, err)
return
}
r.incoming <- nitroMsg
default:
r.Logger.Error(fmt.Sprintf("Unknown message type: %T", e.Message))
}
default:
r.Logger.Error(fmt.Sprintf("Unknown channel ID: %X", e.ChannelID))
}
}
func (r *p2pReactor) OnStart() error {
// Start the single sending goroutine
r.wg.Add(1)
go r.sendLoop()
return nil
}
func (r *p2pReactor) OnStop() { r.close() }
func (r *p2pReactor) close() {
close(r.incoming)
close(r.sendQueue)
r.wg.Wait() // Wait for sending goroutine to finish
r.mtx.Lock()
defer r.mtx.Unlock()
for id, ch := range r.outgoing {
close(ch)
delete(r.outgoing, id)
}
}
// sendLoop runs in a single goroutine and handles all outgoing messages with rate limiting
func (r *p2pReactor) sendLoop() {
defer r.wg.Done()
for msg := range r.sendQueue {
// Apply rate limiting if enabled
if r.rateLimiter != nil {
ctx, cancel := context.WithTimeout(context.Background(), limiterTimeout)
if err := r.rateLimiter.Wait(ctx); err != nil {
r.logger.Error("Rate limit exceeded, dropping message", "error", err)
cancel()
continue
}
cancel()
}
// Send to all connected peers
r.mtx.RLock()
for peerID, ch := range r.outgoing {
select {
case ch <- msg:
// Message sent successfully
default:
// Channel is full, log warning but don't block
r.logger.Warn("Peer channel full, dropping message", "peer", peerID)
}
}
r.mtx.RUnlock()
}
}
func (r *p2pReactor) send(msg nitrop2p.Message) {
select {
case r.sendQueue <- msg:
// Message queued successfully
default:
// Send queue is full, log error
r.logger.Error("Send queue full, dropping message")
}
}
// GetRateLimitStats returns current rate limiting statistics
func (r *p2pReactor) GetRateLimitStats() (enabled bool, limit float64, burst int, tokens float64) {
if r.rateLimiter == nil {
return false, 0, 0, 0
}
return true, float64(r.rateLimiter.Limit()), r.rateLimiter.Burst(), r.rateLimiter.Tokens()
}
// UpdateRateLimit updates the rate limiting parameters at runtime
func (r *p2pReactor) UpdateRateLimit(newRate float64, newBurst int) {
if r.rateLimiter != nil {
r.rateLimiter.SetLimit(rate.Limit(newRate))
r.rateLimiter.SetBurst(newBurst)
}
}
// MessageService
func newMessageService(id string, r *p2pReactor /* , logger log.Logger */) *p2pMsgService {
return &p2pMsgService{
id: id,
reactor: r,
// logger: logger,
}
}
func (ms *p2pMsgService) P2PMessages() <-chan nitrop2p.Message {
return ms.reactor.incoming
}
func (ms *p2pMsgService) Send(p nitrotypes.Participant, m nitrop2p.Message) error {
if !ms.reactor.IsRunning() {
return errors.New("reactor not running")
}
ms.reactor.send(m)
return nil
}
func (ms *p2pMsgService) Close() error {
// the lifecycle of the reactor is managed by cometbft
return nil
}
func (ms *p2pMsgService) Id() string {
return ms.id
}
server/nitro/p2p.pb.go
+317
View File
@@ -0,0 +1,317 @@
// Code generated by protoc-gen-gogo. DO NOT EDIT.
// source: cerc/nitro/v1/p2p.proto
package nitro
import (
fmt "fmt"
proto "github.com/cosmos/gogoproto/proto"
io "io"
math "math"
math_bits "math/bits"
)
// Reference imports to suppress errors if they are not otherwise used.
var _ = proto.Marshal
var _ = fmt.Errorf
var _ = math.Inf
// This is a compile-time assertion to ensure that this generated file
// is compatible with the proto package it is being compiled against.
// A compilation error at this line likely means your copy of the
// proto package needs to be updated.
const _ = proto.GoGoProtoPackageIsVersion3 // please upgrade the proto package
// P2PMessage wraps a json-encoded Nitro protocols.Message
// TODO: proper proto message?
type P2PMessage struct {
Content string `protobuf:"bytes,1,opt,name=content,proto3" json:"content,omitempty"`
}
func (m *P2PMessage) Reset() { *m = P2PMessage{} }
func (m *P2PMessage) String() string { return proto.CompactTextString(m) }
func (*P2PMessage) ProtoMessage() {}
func (*P2PMessage) Descriptor() ([]byte, []int) {
return fileDescriptor_97350824263b9206, []int{0}
}
func (m *P2PMessage) XXX_Unmarshal(b []byte) error {
return m.Unmarshal(b)
}
func (m *P2PMessage) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
if deterministic {
return xxx_messageInfo_P2PMessage.Marshal(b, m, deterministic)
} else {
b = b[:cap(b)]
n, err := m.MarshalToSizedBuffer(b)
if err != nil {
return nil, err
}
return b[:n], nil
}
}
func (m *P2PMessage) XXX_Merge(src proto.Message) {
xxx_messageInfo_P2PMessage.Merge(m, src)
}
func (m *P2PMessage) XXX_Size() int {
return m.Size()
}
func (m *P2PMessage) XXX_DiscardUnknown() {
xxx_messageInfo_P2PMessage.DiscardUnknown(m)
}
var xxx_messageInfo_P2PMessage proto.InternalMessageInfo
func (m *P2PMessage) GetContent() string {
if m != nil {
return m.Content
}
return ""
}
func init() {
proto.RegisterType((*P2PMessage)(nil), "cerc.nitro.v1.P2PMessage")
}
func init() { proto.RegisterFile("cerc/nitro/v1/p2p.proto", fileDescriptor_97350824263b9206) }
var fileDescriptor_97350824263b9206 = []byte{
// 162 bytes of a gzipped FileDescriptorProto
0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xe2, 0x12, 0x4f, 0x4e, 0x2d, 0x4a,
0xd6, 0xcf, 0xcb, 0x2c, 0x29, 0xca, 0xd7, 0x2f, 0x33, 0xd4, 0x2f, 0x30, 0x2a, 0xd0, 0x2b, 0x28,
0xca, 0x2f, 0xc9, 0x17, 0xe2, 0x05, 0x49, 0xe8, 0x81, 0x25, 0xf4, 0xca, 0x0c, 0x95, 0xd4, 0xb8,
0xb8, 0x02, 0x8c, 0x02, 0x7c, 0x53, 0x8b, 0x8b, 0x13, 0xd3, 0x53, 0x85, 0x24, 0xb8, 0xd8, 0x93,
0xf3, 0xf3, 0x4a, 0x52, 0xf3, 0x4a, 0x24, 0x18, 0x15, 0x18, 0x35, 0x38, 0x83, 0x60, 0x5c, 0x27,
0xa7, 0x13, 0x8f, 0xe4, 0x18, 0x2f, 0x3c, 0x92, 0x63, 0x7c, 0xf0, 0x48, 0x8e, 0x71, 0xc2, 0x63,
0x39, 0x86, 0x0b, 0x8f, 0xe5, 0x18, 0x6e, 0x3c, 0x96, 0x63, 0x88, 0xd2, 0x48, 0xcf, 0x2c, 0xd1,
0x2b, 0x4b, 0x49, 0xd2, 0x2b, 0xc9, 0xd7, 0x07, 0x99, 0xad, 0x9b, 0x99, 0xaf, 0x9f, 0x93, 0x98,
0x9c, 0x9f, 0x97, 0x99, 0x9c, 0xa2, 0x5f, 0x9c, 0x5a, 0x54, 0x96, 0x5a, 0x04, 0x71, 0x47, 0x12,
0x1b, 0xd8, 0x05, 0xc6, 0x80, 0x00, 0x00, 0x00, 0xff, 0xff, 0x79, 0x5d, 0x6d, 0x7c, 0x9c, 0x00,
0x00, 0x00,
}
func (m *P2PMessage) Marshal() (dAtA []byte, err error) {
size := m.Size()
dAtA = make([]byte, size)
n, err := m.MarshalToSizedBuffer(dAtA[:size])
if err != nil {
return nil, err
}
return dAtA[:n], nil
}
func (m *P2PMessage) MarshalTo(dAtA []byte) (int, error) {
size := m.Size()
return m.MarshalToSizedBuffer(dAtA[:size])
}
func (m *P2PMessage) MarshalToSizedBuffer(dAtA []byte) (int, error) {
i := len(dAtA)
_ = i
var l int
_ = l
if len(m.Content) > 0 {
i -= len(m.Content)
copy(dAtA[i:], m.Content)
i = encodeVarintP2P(dAtA, i, uint64(len(m.Content)))
i--
dAtA[i] = 0xa
}
return len(dAtA) - i, nil
}
func encodeVarintP2P(dAtA []byte, offset int, v uint64) int {
offset -= sovP2P(v)
base := offset
for v >= 1<<7 {
dAtA[offset] = uint8(v&0x7f | 0x80)
v >>= 7
offset++
}
dAtA[offset] = uint8(v)
return base
}
func (m *P2PMessage) Size() (n int) {
if m == nil {
return 0
}
var l int
_ = l
l = len(m.Content)
if l > 0 {
n += 1 + l + sovP2P(uint64(l))
}
return n
}
func sovP2P(x uint64) (n int) {
return (math_bits.Len64(x|1) + 6) / 7
}
func sozP2P(x uint64) (n int) {
return sovP2P(uint64((x << 1) ^ uint64((int64(x) >> 63))))
}
func (m *P2PMessage) Unmarshal(dAtA []byte) error {
l := len(dAtA)
iNdEx := 0
for iNdEx < l {
preIndex := iNdEx
var wire uint64
for shift := uint(0); ; shift += 7 {
if shift >= 64 {
return ErrIntOverflowP2P
}
if iNdEx >= l {
return io.ErrUnexpectedEOF
}
b := dAtA[iNdEx]
iNdEx++
wire |= uint64(b&0x7F) << shift
if b < 0x80 {
break
}
}
fieldNum := int32(wire >> 3)
wireType := int(wire & 0x7)
if wireType == 4 {
return fmt.Errorf("proto: P2PMessage: wiretype end group for non-group")
}
if fieldNum <= 0 {
return fmt.Errorf("proto: P2PMessage: illegal tag %d (wire type %d)", fieldNum, wire)
}
switch fieldNum {
case 1:
if wireType != 2 {
return fmt.Errorf("proto: wrong wireType = %d for field Content", wireType)
}
var stringLen uint64
for shift := uint(0); ; shift += 7 {
if shift >= 64 {
return ErrIntOverflowP2P
}
if iNdEx >= l {
return io.ErrUnexpectedEOF
}
b := dAtA[iNdEx]
iNdEx++
stringLen |= uint64(b&0x7F) << shift
if b < 0x80 {
break
}
}
intStringLen := int(stringLen)
if intStringLen < 0 {
return ErrInvalidLengthP2P
}
postIndex := iNdEx + intStringLen
if postIndex < 0 {
return ErrInvalidLengthP2P
}
if postIndex > l {
return io.ErrUnexpectedEOF
}
m.Content = string(dAtA[iNdEx:postIndex])
iNdEx = postIndex
default:
iNdEx = preIndex
skippy, err := skipP2P(dAtA[iNdEx:])
if err != nil {
return err
}
if (skippy < 0) || (iNdEx+skippy) < 0 {
return ErrInvalidLengthP2P
}
if (iNdEx + skippy) > l {
return io.ErrUnexpectedEOF
}
iNdEx += skippy
}
}
if iNdEx > l {
return io.ErrUnexpectedEOF
}
return nil
}
func skipP2P(dAtA []byte) (n int, err error) {
l := len(dAtA)
iNdEx := 0
depth := 0
for iNdEx < l {
var wire uint64
for shift := uint(0); ; shift += 7 {
if shift >= 64 {
return 0, ErrIntOverflowP2P
}
if iNdEx >= l {
return 0, io.ErrUnexpectedEOF
}
b := dAtA[iNdEx]
iNdEx++
wire |= (uint64(b) & 0x7F) << shift
if b < 0x80 {
break
}
}
wireType := int(wire & 0x7)
switch wireType {
case 0:
for shift := uint(0); ; shift += 7 {
if shift >= 64 {
return 0, ErrIntOverflowP2P
}
if iNdEx >= l {
return 0, io.ErrUnexpectedEOF
}
iNdEx++
if dAtA[iNdEx-1] < 0x80 {
break
}
}
case 1:
iNdEx += 8
case 2:
var length int
for shift := uint(0); ; shift += 7 {
if shift >= 64 {
return 0, ErrIntOverflowP2P
}
if iNdEx >= l {
return 0, io.ErrUnexpectedEOF
}
b := dAtA[iNdEx]
iNdEx++
length |= (int(b) & 0x7F) << shift
if b < 0x80 {
break
}
}
if length < 0 {
return 0, ErrInvalidLengthP2P
}
iNdEx += length
case 3:
depth++
case 4:
if depth == 0 {
return 0, ErrUnexpectedEndOfGroupP2P
}
depth--
case 5:
iNdEx += 4
default:
return 0, fmt.Errorf("proto: illegal wireType %d", wireType)
}
if iNdEx < 0 {
return 0, ErrInvalidLengthP2P
}
if depth == 0 {
return iNdEx, nil
}
}
return 0, io.ErrUnexpectedEOF
}
var (
ErrInvalidLengthP2P = fmt.Errorf("proto: negative length found during unmarshaling")
ErrIntOverflowP2P = fmt.Errorf("proto: integer overflow")
ErrUnexpectedEndOfGroupP2P = fmt.Errorf("proto: unexpected end of group")
)
server/nitro/p2p_ratelimit_test.go
+74
View File
@@ -0,0 +1,74 @@
package nitro
import (
"testing"
"cosmossdk.io/log"
nitrop2p "github.com/statechannels/go-nitro/node/engine/messageservice"
nitrotypes "github.com/statechannels/go-nitro/types"
"github.com/stretchr/testify/assert"
)
func TestP2PRateLimit(t *testing.T) {
// Test with rate limiting enabled
config := &Config{
P2PRateLimitEnable: true,
P2PRateLimitRate: 2.0, // 2 messages per second
P2PRateLimitBurst: 2, // burst of 2
}
reactor := newReactor(config, log.NewNopLogger())
// Check rate limit stats
enabled, limit, burst, tokens := reactor.GetRateLimitStats()
assert.True(t, enabled, "Rate limiting should be enabled")
assert.Equal(t, 2.0, limit, "Rate limit should be 2.0")
assert.Equal(t, 2, burst, "Burst should be 2")
assert.Equal(t, float64(2), tokens, "Initial tokens should equal burst size")
// Test that UpdateRateLimit works
reactor.UpdateRateLimit(5.0, 10)
enabled, limit, burst, _ = reactor.GetRateLimitStats()
assert.True(t, enabled)
assert.Equal(t, 5.0, limit)
assert.Equal(t, 10, burst)
}
func TestP2PRateLimitDisabled(t *testing.T) {
// Test with rate limiting disabled
config := &Config{
P2PRateLimitEnable: false,
}
reactor := newReactor(config, log.NewNopLogger())
// Check rate limit stats
enabled, limit, burst, tokens := reactor.GetRateLimitStats()
assert.False(t, enabled, "Rate limiting should be disabled")
assert.Equal(t, 0.0, limit)
assert.Equal(t, 0, burst)
assert.Equal(t, 0.0, tokens)
}
func TestP2PRateLimitSendBehavior(t *testing.T) {
// Test actual send behavior with rate limiting
config := &Config{
P2PRateLimitEnable: true,
P2PRateLimitRate: 1000.0, // High rate for testing
P2PRateLimitBurst: 1, // Small burst
}
reactor := newReactor(config, log.NewNopLogger())
msgService := newMessageService("test", reactor)
// Create a dummy participant and message
participant, _ := nitrotypes.ParseParticipant("0x0000000000000000000000000000000000000001")
message := nitrop2p.Message{}
// This should return an error because reactor is not running
err := msgService.Send(participant, message)
// We expect an error because reactor is not running
assert.Error(t, err)
assert.Contains(t, err.Error(), "reactor not running")
}
server/nitro/protocol.go
+166
View File
@@ -0,0 +1,166 @@
package nitro
import (
"errors"
"fmt"
"math/big"
"cosmossdk.io/math"
sdk "github.com/cosmos/cosmos-sdk/types"
"github.com/statechannels/go-nitro/channel/state/outcome"
"github.com/statechannels/go-nitro/protocols"
nitrotypes "github.com/statechannels/go-nitro/types"
"git.vdb.to/cerc-io/laconicd/x/nitro/types/v1"
)
type MsgWrapper struct {
Msg sdk.Msg
Objective protocols.ObjectiveId
SignerField *string
}
func (s *Server) OpenLedgerChannel(funds nitrotypes.Funds, counterparty nitrotypes.Participant) (*MsgWrapper, error) {
// TODO: handle multiple assets
if len(funds) != 1 {
return nil, errors.New("only one asset is supported")
}
fund, exit := makeOutcome(s.ethAddress, counterparty, funds)
r, err := s.node.CreateLedgerChannel(counterparty, 0, exit)
if err != nil {
return nil, err
}
s.logger.Info("Initiated objective", "id", r.Id)
// Convert nitrotypes.Funds back to sdk.Coins for the message
// Use "token" as a default denomination for now
// TODO: nitro will store a mapping of denoms to token addresses
// an entry will be added for new denominations upon confirmation of ledger deposit
coins := sdk.Coins{sdk.NewCoin("token", math.NewIntFromBigInt(fund.Amount))}
msg := &v1.MsgOpenChannel{
Funds: coins,
NitroAddress: s.ethAddress.String(),
Counterparty: counterparty.String(),
ChannelId: r.ChannelId.String(),
}
return &MsgWrapper{msg, r.Id, &msg.Signer}, nil
}
func (s *Server) CloseLedgerChannel(channelId nitrotypes.ChannelID, isChallenge bool) (*MsgWrapper, error) {
objective, err := s.node.CloseLedgerChannel(channelId, isChallenge)
if err != nil {
return nil, err
}
s.logger.Info("Initiated objective", "id", objective)
msg := &v1.MsgCloseChannel{
ChannelId: channelId.String(),
IsChallenge: isChallenge,
}
return &MsgWrapper{msg, objective, &msg.Signer}, nil
}
func (s *Server) CreatePaymentChannel(intermediaries []string, counterpartyString string, challengeDuration uint32, funds nitrotypes.Funds) (*MsgWrapper, error) {
// Parse intermediaries
var intermediaryParticipants []nitrotypes.Participant
for _, addr := range intermediaries {
participant, err := nitrotypes.ParseParticipant(addr)
if err != nil {
return nil, fmt.Errorf("failed to parse intermediary %s: %w", addr, err)
}
intermediaryParticipants = append(intermediaryParticipants, participant)
}
// Parse counterparty
counterparty, err := nitrotypes.ParseParticipant(counterpartyString)
if err != nil {
return nil, fmt.Errorf("failed to parse counterparty: %w", err)
}
// TODO: handle multiple assets
if len(funds) != 1 {
return nil, errors.New("only one asset is supported")
}
if len(funds) != 1 {
return nil, errors.New("only one asset is supported")
}
fund, exit := makeOutcome(s.ethAddress, counterparty, funds)
// Create payment channel
r, err := s.node.CreatePaymentChannel(intermediaryParticipants, counterparty, challengeDuration, exit)
if err != nil {
return nil, err
}
s.logger.Info("Initiated payment channel objective", "id", r.Id)
// Convert nitrotypes.Funds back to sdk.Coins for the message
coins := sdk.Coins{sdk.NewCoin("token", math.NewIntFromBigInt(fund.Amount))}
msg := &v1.MsgCreatePaymentChannel{
NitroAddress: s.ethAddress.String(),
Counterparty: counterpartyString,
Intermediaries: intermediaries,
ChallengeDuration: challengeDuration,
Funds: coins,
ChannelId: r.ChannelId.String(),
}
return &MsgWrapper{msg, r.Id, &msg.Signer}, nil
}
func (s *Server) ClosePaymentChannel(channelId nitrotypes.ChannelID) (*MsgWrapper, error) {
objective, err := s.node.ClosePaymentChannel(channelId)
if err != nil {
return nil, err
}
s.logger.Info("Initiated payment channel close objective", "id", objective)
msg := &v1.MsgClosePaymentChannel{
ChannelId: channelId.String(),
}
return &MsgWrapper{msg, objective, &msg.Signer}, nil
}
func (s *Server) Pay(channelId nitrotypes.ChannelID, amount sdk.Coin) (*MsgWrapper, error) {
err := s.node.Pay(channelId, amount.Amount.BigInt())
if err != nil {
return nil, err
}
nodeAddrStr, err := s.addressCodec.BytesToString(s.nodeAddress())
if err != nil {
return nil, err
}
msg := &v1.MsgPay{
ChannelId: channelId.String(),
Amount: amount,
Signer: nodeAddrStr,
}
return &MsgWrapper{msg, "", &msg.Signer}, nil
}
func makeOutcome(ethAddress, counterparty nitrotypes.Participant, funds nitrotypes.Funds) (nitrotypes.Fund, outcome.Exit) {
var fund nitrotypes.Fund
for asset, amount := range funds {
fund = nitrotypes.Fund{asset, amount}
}
exit := outcome.Exit{
outcome.SingleAssetExit{
Asset: fund.Asset,
Allocations: []outcome.Allocation{
{
Destination: ethAddress.ToDestination(),
Amount: fund.Amount,
},
{
Destination: counterparty.ToDestination(),
Amount: big.NewInt(0),
},
},
},
}
return fund, exit
}
server/nitro/query.go
+194
View File
@@ -0,0 +1,194 @@
package nitro
import (
"encoding/json"
"fmt"
"github.com/cosmos/cosmos-sdk/client/flags"
"github.com/spf13/cobra"
nitrotypes "github.com/statechannels/go-nitro/types"
"git.vdb.to/cerc-io/laconicd/server"
"git.vdb.to/cerc-io/laconicd/server/relay"
)
func GetChannelCmd() *cobra.Command {
cmd := &cobra.Command{
Use: "get-channel <channel-id>",
Short: "Get information about a channel",
Args: cobra.ExactArgs(1),
Long: "Retrieves detailed information about a specific ledger or payment channel.",
RunE: func(cmd *cobra.Command, args []string) error {
channelId, err := nitrotypes.ParseChannelID(args[0])
if err != nil {
return fmt.Errorf("failed to parse channel ID: %w", err)
}
ledger, _ := cmd.Flags().GetBool("ledger")
payment, _ := cmd.Flags().GetBool("payment")
local, _ := cmd.Flags().GetBool("local")
if !ledger && !payment {
ledger = true
}
if ledger && payment {
return fmt.Errorf("cannot specify both --ledger and --payment flags")
}
s, done, err := setupNitroQueryCommand(cmd, !local)
if err != nil {
return err
}
defer done()
var output []byte
if ledger {
info, err := s.node.GetLedgerChannel(channelId)
if err != nil {
return fmt.Errorf("failed to get ledger channel: %w", err)
}
output, err = json.MarshalIndent(info, "", " ")
if err != nil {
return fmt.Errorf("failed to marshal response: %w", err)
}
} else {
info, err := s.node.GetPaymentChannel(channelId)
if err != nil {
return fmt.Errorf("failed to get payment channel: %w", err)
}
output, err = json.MarshalIndent(info, "", " ")
if err != nil {
return fmt.Errorf("failed to marshal response: %w", err)
}
}
fmt.Println(string(output))
return nil
},
}
cmd.Flags().Bool("ledger", false, "Get ledger channel information")
cmd.Flags().Bool("payment", false, "Get payment channel information")
cmd.Flags().Bool("local", false, "Query local data only (no P2P connection)")
flags.AddQueryFlagsToCmd(cmd)
server.SetRequiredComponents(cmd, &Server{}, &relay.Server{})
return cmd
}
func ListChannelsCmd() *cobra.Command {
cmd := &cobra.Command{
Use: "list-channels",
Short: "List all ledger channels",
Args: cobra.NoArgs,
Long: "Retrieves information about all ledger channels.",
RunE: func(cmd *cobra.Command, args []string) error {
local, _ := cmd.Flags().GetBool("local")
s, done, err := setupNitroQueryCommand(cmd, !local)
if err != nil {
return err
}
defer done()
channels, err := s.node.GetAllLedgerChannels()
if err != nil {
return fmt.Errorf("failed to get all ledger channels: %w", err)
}
output, err := json.MarshalIndent(channels, "", " ")
if err != nil {
return fmt.Errorf("failed to marshal response: %w", err)
}
fmt.Println(string(output))
return nil
},
}
cmd.Flags().Bool("local", false, "Query local data only (no P2P connection)")
flags.AddQueryFlagsToCmd(cmd)
server.SetRequiredComponents(cmd, &Server{}, &relay.Server{})
return cmd
}
func ListPaymentChannelsCmd() *cobra.Command {
cmd := &cobra.Command{
Use: "list-payment-channels <ledger-channel-id>",
Short: "List all payment channels for a ledger channel",
Args: cobra.ExactArgs(1),
Long: "Retrieves all active payment channels associated with a specific ledger channel.",
RunE: func(cmd *cobra.Command, args []string) error {
ledgerChannelId, err := nitrotypes.ParseChannelID(args[0])
if err != nil {
return fmt.Errorf("failed to parse ledger channel ID: %w", err)
}
local, _ := cmd.Flags().GetBool("local")
s, done, err := setupNitroQueryCommand(cmd, !local)
if err != nil {
return err
}
defer done()
channels, err := s.node.GetPaymentChannelsByLedger(ledgerChannelId)
if err != nil {
return fmt.Errorf("failed to get payment channels: %w", err)
}
output, err := json.MarshalIndent(channels, "", " ")
if err != nil {
return fmt.Errorf("failed to marshal response: %w", err)
}
fmt.Println(string(output))
return nil
},
}
cmd.Flags().Bool("local", false, "Query local data only (no P2P connection)")
flags.AddQueryFlagsToCmd(cmd)
server.SetRequiredComponents(cmd, &Server{}, &relay.Server{})
return cmd
}
func IdentityCmd() *cobra.Command {
cmd := &cobra.Command{
Use: "identity",
Short: "Get the node identity",
Args: cobra.NoArgs,
Long: "Retrieves the node's identity. Use --participant for principal ID (default) or --network for network address.",
RunE: func(cmd *cobra.Command, args []string) error {
participant, _ := cmd.Flags().GetBool("participant")
network, _ := cmd.Flags().GetBool("network")
if !participant && !network {
participant = true
}
if participant && network {
return fmt.Errorf("cannot specify both --participant and --network flags")
}
s, done, err := setupNitroQueryCommand(cmd, false)
if err != nil {
return err
}
defer done()
if participant {
identity := s.node.PrincipalID()
fmt.Println(identity.String())
} else {
identity := s.node.NetworkID()
fmt.Println(identity.String())
}
return nil
},
}
cmd.Flags().Bool("participant", false, "Get the principal ID (default)")
cmd.Flags().Bool("network", false, "Get the network address")
flags.AddQueryFlagsToCmd(cmd)
server.SetRequiredComponents(cmd, &Server{})
return cmd
}
server/nitro/server.go
+210
View File
@@ -0,0 +1,210 @@
package nitro
import (
"context"
"encoding/hex"
"fmt"
"log/slog"
"path/filepath"
"sync/atomic"
cmtcrypto "github.com/cometbft/cometbft/crypto"
"github.com/cometbft/cometbft/p2p"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/crypto"
"github.com/spf13/pflag"
nitrocrypto "github.com/statechannels/go-nitro/crypto"
"github.com/statechannels/go-nitro/node"
"github.com/statechannels/go-nitro/node/engine"
"github.com/statechannels/go-nitro/node/engine/chainservice"
chainutils "github.com/statechannels/go-nitro/node/engine/chainservice/utils"
"github.com/statechannels/go-nitro/node/engine/store"
nitrotypes "github.com/statechannels/go-nitro/types"
"cosmossdk.io/core/address"
"cosmossdk.io/log"
"github.com/cosmos/cosmos-sdk/client/flags"
"github.com/cosmos/cosmos-sdk/crypto/keyring"
sdk "github.com/cosmos/cosmos-sdk/types"
"git.vdb.to/cerc-io/laconicd/server"
"git.vdb.to/cerc-io/laconicd/utils"
)
const (
serverName = "nitro"
ServerContextKey = "server." + serverName
)
// TODO
// 1. implement participation as a signing group member:
// 1.1. initialize group delegate credentials, with distsig over ABCI
// 1.2. bind custodian contract for chain/adjudicator actions
// 2. operation to update DKG group members
// 3. integrate with onboarding module
// Server wraps and configures a go-nitro Node.
type Server struct {
logger log.Logger
config *Config
node *node.Node
storeDir string // path to Nitro store directory
reactor *p2pReactor
keyring keyring.Keyring
addressCodec address.Codec
ethAddress nitrotypes.Address
nodeKey cmtcrypto.PrivKey
// participantID nitrotypes.Participant
ready atomic.Bool
}
func New(globalConfig server.ConfigMap, logger log.Logger, kr keyring.Keyring) (*Server, error) {
home, _ := globalConfig[flags.FlagHome].(string)
cfg, err := UnmarshalConfig(globalConfig)
if err != nil {
return nil, err
}
if err := cfg.Validate(); err != nil {
return nil, err
}
s := &Server{
config: cfg,
storeDir: filepath.Join(home, "nitro"),
reactor: newReactor(cfg, logger.With(log.ModuleKey, serverName, "component", "p2p-reactor")),
keyring: kr,
logger: logger.With(log.ModuleKey, serverName),
addressCodec: utils.NewAddressCodec(),
}
return s, nil
}
func (*Server) Name() string {
return serverName
}
func (s *Server) Start(ctx context.Context) error {
if !s.config.Enable {
s.logger.Info(fmt.Sprintf("%s server is disabled via config", s.Name()))
return nil
}
// TODO: pass keyring through context?
if err := s.init(s.keyring); err != nil {
return err
}
s.logger.Info("starting Nitro server")
s.ready.Store(true)
return nil
}
func (s *Server) Stop(context.Context) error {
s.logger.Info("stopping Nitro server")
s.ready.Store(false)
if s.node != nil {
return s.node.Close()
}
return nil
}
func (s *Server) init(kr keyring.Keyring) error {
var (
ethkey cmtcrypto.PrivKey
err error
c = s.config
)
if c.EthKey != "" {
ethkey, err = utils.ExtractPrivateKeyByUid(kr, c.EthKey)
if err != nil {
return err
}
} else {
return fmt.Errorf("eth-key is not set") // should already be caught by Config.Validate
}
s.nodeKey = ethkey
storeOpts := store.StoreOpts{
UseDurableStore: true,
DurableStoreDir: s.storeDir,
}
chainOpts := chainservice.ChainOpts{
ChainUrl: c.EthURL,
StartBlockNum: c.EthStartBlock,
ChainAuthToken: c.EthAuthToken,
ChainPk: hex.EncodeToString(ethkey.Bytes()),
}
// inject SDK logger into slog, which Nitro uses
loggerImpl, ok := s.logger.Impl().(*slog.Logger)
if ok {
slog.SetDefault(loggerImpl)
} else {
s.logger.Warn("Logger backend is not slog, cannot set as Nitro logger")
}
store, err := store.NewStore(storeOpts)
if err != nil {
return err
}
// Compare chainOpts.ChainStartBlock to lastBlockNum seen in store. The larger of the two
// gets passed as an argument when creating NewEthChainService
storeBlockNum, err := store.GetLastBlockNumSeen()
if err != nil {
return err
}
chainOpts.StartBlockNum = max(storeBlockNum, chainOpts.StartBlockNum)
contractAddresses := node.ContractAddresses{
NaAddress: common.HexToAddress(c.EthNaAddress),
VpaAddress: common.HexToAddress(c.EthVpaAddress),
CaAddress: common.HexToAddress(c.EthCaAddress),
}
chainOpts.CreateAdjudicator = chainutils.AdjudicatorCreator(contractAddresses.NaAddress)
chain, err := chainservice.NewL1ChainService(chainOpts)
if err != nil {
return err
}
// note: utils.EthAddressFromPubKey expects uncompressed keys
ethPubkey, err := crypto.DecompressPubkey(ethkey.PubKey().Bytes())
if err != nil {
return err
}
s.ethAddress = nitrotypes.Address(crypto.PubkeyToAddress(*ethPubkey))
s.node = node.New(
newMessageService("laconicd-p2p-"+s.ethAddress.String(), s.reactor),
chain,
store,
&engine.PermissivePolicy{},
nitrocrypto.NewSimpleCredential(ethkey.Bytes()),
contractAddresses,
)
return nil
}
func (s *Server) Config() any {
if s.config == nil {
return DefaultConfig()
}
return s.config
}
func (s *Server) StartCmdFlags() *pflag.FlagSet {
flags := pflag.NewFlagSet(s.Name(), pflag.ExitOnError)
AddFlags(flags)
return flags
}
func (s *Server) Ready() bool {
return s.ready.Load()
}
func (s *Server) ContextKey() string {
return ServerContextKey
}
func (s *Server) P2PReactors() map[string]p2p.Reactor {
return map[string]p2p.Reactor{"NITRO": s.reactor}
}
func (s *Server) nodeAddress() sdk.AccAddress {
return sdk.AccAddress(s.nodeKey.PubKey().Address())
}
server/nitro/tx.go
+326
View File
@@ -0,0 +1,326 @@
package nitro
import (
"encoding/json"
"fmt"
"math/big"
"github.com/cosmos/cosmos-sdk/client/flags"
sdk "github.com/cosmos/cosmos-sdk/types"
"github.com/spf13/cobra"
"github.com/statechannels/go-nitro/node/query"
nitrotypes "github.com/statechannels/go-nitro/types"
"git.vdb.to/cerc-io/laconicd/server"
"git.vdb.to/cerc-io/laconicd/server/relay"
)
func OpenChannelCmd() *cobra.Command {
cmd := &cobra.Command{
Use: "open-channel <counterparty> <amount>",
Short: "Open a ledger channel with the given counterparty",
Args: cobra.ExactArgs(2),
Long: "Creates a ledger channel with the specified counterparty and funding amount.",
RunE: func(cmd *cobra.Command, args []string) error {
counterparty, err := nitrotypes.ParseParticipant(args[0])
if err != nil {
return fmt.Errorf("failed to parse counterparty: %w", err)
}
clientCtx, s, done, err := setupNitroCommand(cmd)
if err != nil {
return err
}
defer done()
assetAddr, _ := cmd.Flags().GetString("asset")
funds, err := resolveTokens(s, args[1], assetAddr)
if err != nil {
return err
}
m, err := s.OpenLedgerChannel(funds, counterparty)
if err != nil {
return err
}
if err = waitForObjective(
s.node.FailedObjectives(),
m.Objective,
s.node.LedgerUpdatedChan(m.Objective.ChannelID()),
func(info query.LedgerChannelInfo) bool { return info.Status == query.Open },
); err != nil {
return err
}
return submitNitroTx(*clientCtx, cmd, *m, false)
},
}
cmd.Flags().String("asset", "", "Ethereum token address (if specified, amount is parsed as decimal without denomination)")
AddFlags(cmd.Flags())
flags.AddTxFlagsToCmd(cmd)
server.SetRequiredComponents(cmd, &Server{}, &relay.Server{})
return cmd
}
func CloseChannelCmd() *cobra.Command {
cmd := &cobra.Command{
Use: "close-channel <channel-id>",
Short: "Close a ledger channel",
Args: cobra.ExactArgs(1),
Long: "Close a ledger channel. Use --challenge flag for dispute resolution when cooperation fails.",
RunE: func(cmd *cobra.Command, args []string) error {
channelId, err := nitrotypes.ParseChannelID(args[0])
if err != nil {
return fmt.Errorf("failed to parse channel ID: %w", err)
}
isChallenge, err := cmd.Flags().GetBool("challenge")
if err != nil {
return err
}
clientCtx, s, done, err := setupNitroCommand(cmd)
if err != nil {
return err
}
defer done()
m, err := s.CloseLedgerChannel(channelId, isChallenge)
if err != nil {
return err
}
if err = waitForObjective(
s.node.FailedObjectives(),
m.Objective,
s.node.LedgerUpdatedChan(channelId),
func(info query.LedgerChannelInfo) bool { return info.Status == query.Complete },
); err != nil {
return err
}
return submitNitroTx(*clientCtx, cmd, *m, false)
},
}
cmd.Flags().Bool("challenge", false, "Close via challenge (dispute resolution) instead of cooperative closure")
AddFlags(cmd.Flags())
flags.AddTxFlagsToCmd(cmd)
server.SetRequiredComponents(cmd, &Server{}, &relay.Server{})
return cmd
}
func OpenPaymentChannelCmd() *cobra.Command {
cmd := &cobra.Command{
Use: "open-payment-channel <counterparty> <amount> [intermediaries...]",
Short: "Open a virtual payment channel",
Args: cobra.MinimumNArgs(2),
Long: "Creates a virtual payment channel with the specified counterparty and optional intermediaries.",
RunE: func(cmd *cobra.Command, args []string) error {
counterparty := args[0]
challengeDuration, err := cmd.Flags().GetUint32("challenge-duration")
if err != nil {
return err
}
var intermediaries []string
if len(args) > 2 {
intermediaries = args[2:]
}
clientCtx, s, done, err := setupNitroCommand(cmd)
if err != nil {
return err
}
defer done()
assetAddr, _ := cmd.Flags().GetString("asset")
funds, err := resolveTokens(s, args[1], assetAddr)
if err != nil {
return err
}
m, err := s.CreatePaymentChannel(intermediaries, counterparty, challengeDuration, funds)
if err != nil {
return err
}
if err = waitForObjective(
s.node.FailedObjectives(),
m.Objective,
s.node.PaymentChannelUpdatedChan(m.Objective.ChannelID()),
func(info query.PaymentChannelInfo) bool { return info.Status == query.Open },
); err != nil {
return err
}
return submitNitroTx(*clientCtx, cmd, *m, false)
},
}
cmd.Flags().String("asset", "", "Ethereum token address (if specified, amount is parsed as decimal without denomination)")
cmd.Flags().Uint32("challenge-duration", 0, "Challenge duration for the channel")
AddFlags(cmd.Flags())
flags.AddTxFlagsToCmd(cmd)
server.SetRequiredComponents(cmd, &Server{}, &relay.Server{})
return cmd
}
func ClosePaymentChannelCmd() *cobra.Command {
cmd := &cobra.Command{
Use: "close-payment-channel <channel-id>",
Short: "Close a virtual payment channel",
Args: cobra.ExactArgs(1),
Long: "Close a virtual payment channel with the given channel ID.",
RunE: func(cmd *cobra.Command, args []string) error {
channelId, err := nitrotypes.ParseChannelID(args[0])
if err != nil {
return fmt.Errorf("failed to parse channel ID: %w", err)
}
clientCtx, s, done, err := setupNitroCommand(cmd)
if err != nil {
return err
}
defer done()
m, err := s.ClosePaymentChannel(channelId)
if err != nil {
return err
}
if err = waitForObjective(
s.node.FailedObjectives(),
m.Objective,
s.node.PaymentChannelUpdatedChan(channelId),
func(info query.PaymentChannelInfo) bool { return info.Status == query.Complete },
); err != nil {
return err
}
return submitNitroTx(*clientCtx, cmd, *m, false)
},
}
AddFlags(cmd.Flags())
flags.AddTxFlagsToCmd(cmd)
server.SetRequiredComponents(cmd, &Server{}, &relay.Server{})
return cmd
}
func PayCmd() *cobra.Command {
cmd := &cobra.Command{
Use: "pay <channel-id> <amount>",
Short: "Send a payment through a payment channel",
Args: cobra.ExactArgs(2),
Long: "Creates and sends a payment voucher through the specified payment channel.",
RunE: func(cmd *cobra.Command, args []string) error {
channelId, err := nitrotypes.ParseChannelID(args[0])
if err != nil {
return fmt.Errorf("failed to parse channel ID: %w", err)
}
coins, err := sdk.ParseCoinsNormalized(args[1])
if err != nil {
return fmt.Errorf("failed to parse amount: %w", err)
}
if len(coins) != 1 {
return fmt.Errorf("exactly one coin denomination required")
}
clientCtx, s, done, err := setupNitroCommand(cmd)
if err != nil {
return err
}
defer done()
info, err := s.node.GetPaymentChannel(channelId)
if err != nil {
return err
}
paidSoFar := info.Balance.PaidSoFar.ToInt()
expectedAmount := new(big.Int).Add(paidSoFar, coins[0].Amount.BigInt())
updateChan := s.node.PaymentChannelUpdatedChan(channelId)
m, err := s.Pay(channelId, coins[0])
if err != nil {
return fmt.Errorf("failed to send payment: %w", err)
}
// TODO: unordered tx for payments: this could break if tx is sent before opening the
// channel; think through wrt. state integration
if err = submitNitroTx(*clientCtx, cmd, *m, true); err != nil {
return err
}
for info := range updateChan {
if expectedAmount.Cmp(info.Balance.PaidSoFar.ToInt()) <= 0 {
break
}
}
delayAfterUpdate()
return nil
},
}
AddFlags(cmd.Flags())
flags.AddTxFlagsToCmd(cmd)
server.SetRequiredComponents(cmd, &Server{}, &relay.Server{})
return cmd
}
func CreateVoucherCmd() *cobra.Command {
cmd := &cobra.Command{
Use: "create-voucher <channel-id> <amount>",
Short: "Create a payment voucher for a payment channel",
Args: cobra.ExactArgs(2),
Long: "Creates a signed payment voucher that can be used to redeem payments from the specified channel.",
RunE: func(cmd *cobra.Command, args []string) error {
channelId, err := nitrotypes.ParseChannelID(args[0])
if err != nil {
return fmt.Errorf("failed to parse channel ID: %w", err)
}
coins, err := sdk.ParseCoinsNormalized(args[1])
if err != nil {
return fmt.Errorf("failed to parse amount: %w", err)
}
if len(coins) != 1 {
return fmt.Errorf("exactly one coin denomination required")
}
amount := coins[0].Amount.BigInt()
s, done, err := setupNitroQueryCommand(cmd, false)
if err != nil {
return err
}
defer done()
voucher, err := s.node.CreateVoucher(channelId, amount)
if err != nil {
return fmt.Errorf("failed to create voucher: %w", err)
}
output, err := json.MarshalIndent(voucher, "", " ")
if err != nil {
return fmt.Errorf("failed to marshal voucher: %w", err)
}
fmt.Println(string(output))
return nil
},
}
flags.AddQueryFlagsToCmd(cmd)
server.SetRequiredComponents(cmd, &Server{}, &relay.Server{})
return cmd
}
server/relay/config.go
+23
View File
@@ -0,0 +1,23 @@
package relay
import (
"fmt"
"git.vdb.to/cerc-io/laconicd/server"
)
type Config struct {
// Target string `mapstructure:"target" toml:"target" comment:"The target peer to relay messages to"`
}
func DefaultConfig() *Config {
return &Config{}
}
func UnmarshalConfig(cfg map[string]any) (*Config, error) {
config := DefaultConfig()
if err := server.UnmarshalSubConfig(cfg, serverName, config); err != nil {
return nil, fmt.Errorf("failed to unmarshal %T: %w", config, err)
}
return config, nil
}
server/relay/server.go
+199
View File
@@ -0,0 +1,199 @@
package relay
import (
"context"
"errors"
"fmt"
"strings"
"sync/atomic"
"cosmossdk.io/log"
cmtcfg "github.com/cometbft/cometbft/config"
"github.com/cometbft/cometbft/p2p"
"github.com/cometbft/cometbft/p2p/pex"
"github.com/cometbft/cometbft/version"
cmtversion "github.com/cometbft/cometbft/version"
"git.vdb.to/cerc-io/laconicd/server"
)
const (
serverName = "relay"
ServerContextKey = "server." + serverName
)
// Server is a server component with CometBFT p2p capabilities
type Server struct {
logger log.Logger
config *Config
cometConfig *cmtcfg.Config
nodeKey *p2p.NodeKey
ready atomic.Bool
}
type Switch struct {
*p2p.Switch
transport *p2p.MultiplexTransport
}
func New(cfg server.ConfigMap, logger log.Logger) (*Server, error) {
c, err := UnmarshalConfig(cfg)
if err != nil {
return nil, err
}
cometConfig := cmtcfg.DefaultConfig()
if err := server.UnmarshalSubConfig(cfg, "", &cometConfig); err != nil {
return nil, err
}
return &Server{
config: c,
cometConfig: cometConfig,
logger: logger.With(log.ModuleKey, serverName),
}, nil
}
func (*Server) Name() string {
return serverName
}
func (s *Server) Start(ctx context.Context) error {
var err error
s.nodeKey, err = p2p.LoadOrGenNodeKey(s.cometConfig.NodeKeyFile())
if err != nil {
return err
}
s.ready.Store(true)
return nil
}
func (s *Server) Stop(ctx context.Context) error {
s.ready.Store(false)
return nil
}
func (s *Server) Config() any {
if s.config == nil {
return DefaultConfig()
}
return s.config
}
func (s *Server) Ready() bool {
return s.ready.Load()
}
func (s *Server) ContextKey() string {
return ServerContextKey
}
func (s *Server) GetSwitch(chainID string, blockVersion uint64) (*Switch, error) {
nodeInfo, err := makeNodeInfo(s.cometConfig, s.nodeKey, chainID, blockVersion)
if err != nil {
return nil, err
}
transport := createTransport(s.cometConfig, nodeInfo, s.nodeKey)
sw := p2p.NewSwitch(s.cometConfig.P2P, transport, p2p.SwitchPeerFilters())
sw.SetNodeInfo(nodeInfo)
sw.SetNodeKey(s.nodeKey)
return &Switch{sw, transport}, nil
}
func (s *Switch) AddReactor(name string, reactor p2p.Reactor) error {
s.Switch.AddReactor(name, reactor)
// register the new channels to the nodeInfo
ni, ok := s.NodeInfo().(p2p.DefaultNodeInfo)
if !ok {
return errors.New("Node info is not of type DefaultNodeInfo")
}
for _, chDesc := range reactor.GetChannels() {
if !ni.HasChannel(chDesc.ID) {
ni.Channels = append(ni.Channels, chDesc.ID)
s.transport.AddChannel(chDesc.ID)
}
}
s.SetNodeInfo(ni)
return nil
}
func (s *Switch) Dial(peerAddr string) error {
addr, err := p2p.NewNetAddressString(peerAddr)
if err != nil {
return fmt.Errorf("error resolving net address: %w", err)
}
err = s.AddPersistentPeers([]string{addr.String()})
if err != nil {
return fmt.Errorf("error adding persistent peer: %w", err)
}
return s.DialPeerWithAddress(addr)
}
func makeNodeInfo(
config *cmtcfg.Config,
nodeKey *p2p.NodeKey,
chainID string,
blockVersion uint64,
) (p2p.DefaultNodeInfo, error) {
nodeInfo := p2p.DefaultNodeInfo{
ProtocolVersion: p2p.NewProtocolVersion(
cmtversion.P2PProtocol,
blockVersion,
0,
),
DefaultNodeID: nodeKey.ID(),
Network: chainID,
Version: version.TMCoreSemVer,
Channels: nil, // updated in AddReactor
Moniker: config.Moniker,
Other: p2p.DefaultNodeInfoOther{
RPCAddress: config.RPC.ListenAddress,
},
}
if config.P2P.PexReactor {
nodeInfo.Channels = append(nodeInfo.Channels, pex.PexChannel)
}
lAddr := config.P2P.ExternalAddress
if lAddr == "" {
lAddr = config.P2P.ListenAddress
}
nodeInfo.ListenAddr = lAddr
// err := nodeInfo.Validate()
return nodeInfo, nil
}
func createTransport(
config *cmtcfg.Config,
nodeInfo p2p.NodeInfo,
nodeKey *p2p.NodeKey,
) *p2p.MultiplexTransport {
var (
mConnConfig = p2p.MConnConfig(config.P2P)
transport = p2p.NewMultiplexTransport(nodeInfo, *nodeKey, mConnConfig)
connFilters = []p2p.ConnFilterFunc{}
)
if !config.P2P.AllowDuplicateIP {
connFilters = append(connFilters, p2p.ConnDuplicateIPFilter())
}
p2p.MultiplexTransportConnFilters(connFilters...)(transport)
// Limit the number of incoming connections.
max := config.P2P.MaxNumInboundPeers
p2p.MultiplexTransportMaxIncomingConnections(max)(transport)
return transport
}
func removeProtocolIfDefined(addr string) string {
if strings.Contains(addr, "://") {
return strings.Split(addr, "://")[1]
}
return addr
}
server/relay/util.go
+20
View File
@@ -0,0 +1,20 @@
package relay
import (
"runtime"
"git.vdb.to/cerc-io/laconicd/utils"
"github.com/spf13/cobra"
)
func GetServerFromCmd(cmd *cobra.Command) (*Server, error) {
s, err := utils.GetFromContext[Server](cmd.Context(), ServerContextKey)
if err != nil {
return nil, err
}
for !s.Ready() {
s.logger.Debug("waiting for relay server")
runtime.Gosched()
}
return s, nil
}
server/server.go
+29
View File
@@ -0,0 +1,29 @@
package server
// The utilities in this file are copied from cosmos-sdk server/v2
import (
"context"
"github.com/spf13/pflag"
)
// ServerComponent is a server component that can be started and stopped.
type ServerComponent interface {
// Name returns the name of the server component.
Name() string
// Start starts the server component.
Start(context.Context) error
// Stop stops the server component.
// Once Stop has been called on a server component, it may not be reused.
Stop(context.Context) error
}
// HasStartFlags is a server component that has start flags.
type HasStartFlags interface {
// StartCmdFlags returns server start flags.
// Those flags should be prefixed with the server name.
// They are then merged with the server config in one viper instance.
StartCmdFlags() *pflag.FlagSet
}