You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

353 lines
8.8 KiB

package swarm
import (
"context"
"crypto/ecdsa"
"fmt"
"os"
"os/user"
"strings"
"time"
"github.com/ethereum/go-ethereum/common/hexutil"
"github.com/ethereum/go-ethereum/node"
"github.com/ethereum/go-ethereum/crypto"
"github.com/ethereum/go-ethereum/p2p"
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/p2p/enode"
"github.com/ethereum/go-ethereum/swarm"
swarmapi "github.com/ethereum/go-ethereum/swarm/api"
"github.com/ethereum/go-ethereum/swarm/network"
"github.com/ethereum/go-ethereum/swarm/pss"
)
const (
// MaxPeers is the maximum number of p2p peer connections
MaxPeers = 10
)
// SwarmBootnodes list of bootnodes for the SWARM network
var SwarmBootnodes = []string{
// EF Swarm Bootnode - AWS - eu-central-1
"enode://4c113504601930bf2000c29bcd98d1716b6167749f58bad703bae338332fe93cc9d9204f08afb44100dc7bea479205f5d162df579f9a8f76f8b402d339709023@3.122.203.99:30301",
// EF Swarm Bootnode - AWS - us-west-2
"enode://89f2ede3371bff1ad9f2088f2012984e280287a4e2b68007c2a6ad994909c51886b4a8e9e2ecc97f9910aca538398e0a5804b0ee80a187fde1ba4f32626322ba@52.35.212.179:30301",
}
func newNode(key *ecdsa.PrivateKey, port int, httpport int, wsport int,
datadir string, modules ...string) (*node.Node, *node.Config, error) {
if port == 0 {
port = 30100
}
cfg := &node.DefaultConfig
if key != nil {
cfg.P2P.PrivateKey = key
}
cfg.P2P.MaxPeers = MaxPeers
cfg.P2P.ListenAddr = fmt.Sprintf("0.0.0.0:%d", port)
cfg.P2P.EnableMsgEvents = true
cfg.P2P.NoDiscovery = false
cfg.P2P.DiscoveryV5 = true
cfg.IPCPath = datadir + "/node.ipc"
cfg.DataDir = datadir
if httpport > 0 {
cfg.HTTPHost = node.DefaultHTTPHost
cfg.HTTPPort = httpport
cfg.HTTPCors = []string{"*"}
}
if wsport > 0 {
cfg.WSHost = node.DefaultWSHost
cfg.WSPort = wsport
cfg.WSOrigins = []string{"*"}
for i := 0; i < len(modules); i++ {
cfg.WSModules = append(cfg.WSModules, modules[i])
}
}
stack, err := node.New(cfg)
if err != nil {
return nil, nil, fmt.Errorf("ServiceNode create fail: %v", err)
}
return stack, cfg, nil
}
func newSwarm(privkey *ecdsa.PrivateKey, datadir string, port int) (*swarm.Swarm, *swarmapi.Config, node.ServiceConstructor) {
// create swarm service
swarmCfg := swarmapi.NewConfig()
swarmCfg.SyncEnabled = true
swarmCfg.Port = fmt.Sprintf("%d", port)
swarmCfg.Path = datadir
swarmCfg.HiveParams.Discovery = true
swarmCfg.Discovery = true
swarmCfg.Pss.MsgTTL = time.Second * 10
swarmCfg.Pss.CacheTTL = time.Second * 30
swarmCfg.Pss.AllowRaw = true
swarmCfg.Init(privkey)
swarmNode, err := swarm.NewSwarm(swarmCfg, nil)
if err != nil {
log.Crit("cannot crate swarm node")
}
// register swarm service to the node
var swarmService node.ServiceConstructor = func(ctx *node.ServiceContext) (node.Service, error) {
return swarmNode, nil
}
return swarmNode, swarmCfg, swarmService
}
type swarmPorts struct {
WebSockets int
HTTPRPC int
Bzz int
P2P int
}
func NewSwarmPorts() *swarmPorts {
sp := new(swarmPorts)
sp.WebSockets = 8544
sp.HTTPRPC = 8543
sp.Bzz = 8542
sp.P2P = 31000
return sp
}
type pssSub struct {
Unregister func()
Delivery (chan []byte)
Address string
}
type SwarmNet struct {
Node *node.Node
NodeConfig *node.Config
EnodeID string
Datadir string
Key *ecdsa.PrivateKey
Pss *pss.API
PssPubKey string
PssAddr pss.PssAddress
PssTopics map[string]*pssSub
Hive *network.Hive
Ports *swarmPorts
}
func (sn *SwarmNet) SetLog(level string) error {
// ensure good log formats for terminal
// handle verbosity flag
loglevel, err := log.LvlFromString(level)
if err != nil {
return err
}
hs := log.StreamHandler(os.Stderr, log.TerminalFormat(true))
hf := log.LvlFilterHandler(loglevel, hs)
h := log.CallerFileHandler(hf)
log.Root().SetHandler(h)
return nil
}
func (sn *SwarmNet) PrintStats() {
// statistics thread
go func() {
for {
if sn.Node.Server() != nil && sn.Hive != nil {
addr := fmt.Sprintf("%x", sn.PssAddr)
var addrs [][]byte
addrs = append(addrs, []byte(addr))
peerCount := sn.Node.Server().PeerCount()
log.Info(fmt.Sprintf("PeerCount:%d NeighDepth:%d", peerCount, sn.Hive.NeighbourhoodDepth))
}
time.Sleep(time.Second * 5)
}
}()
}
func (sn *SwarmNet) SetDatadir(datadir string) {
sn.Datadir = datadir
}
func (sn *SwarmNet) SetKey(key *ecdsa.PrivateKey) {
sn.Key = key
}
func (sn *SwarmNet) Init() error {
var err error
if len(sn.Datadir) < 1 {
usr, err := user.Current()
if err != nil {
return err
}
sn.Datadir = usr.HomeDir + "/.dvote/swarm"
os.MkdirAll(sn.Datadir, 0755)
}
sn.SetLog("info")
sn.Ports = NewSwarmPorts()
// create node
sn.Node, sn.NodeConfig, err = newNode(sn.Key, sn.Ports.P2P,
sn.Ports.HTTPRPC, sn.Ports.WebSockets, sn.Datadir, "pss")
if err != nil {
return err
}
// set node key, if not set use the storage one or generate it
if sn.Key == nil {
sn.Key = sn.NodeConfig.NodeKey()
}
// create and register Swarm service
swarmNode, _, swarmHandler := newSwarm(sn.Key, sn.Datadir, sn.Ports.Bzz)
err = sn.Node.Register(swarmHandler)
if err != nil {
return fmt.Errorf("swarm register fail %v", err)
}
// start the node
sn.Node.Start()
for _, url := range SwarmBootnodes {
log.Info("Add bootnode " + url)
node, _ := enode.ParseV4(url)
sn.Node.Server().AddPeer(node)
}
// wait to connect to the p2p network
_, cancel := context.WithTimeout(context.Background(), time.Second)
defer cancel()
time.Sleep(time.Second * 5)
// Get the services API
for _, a := range swarmNode.APIs() {
switch a.Service.(type) {
case *network.Hive:
sn.Hive = a.Service.(*network.Hive)
case *pss.API:
sn.Pss = a.Service.(*pss.API)
}
}
// Create topics map
sn.PssTopics = make(map[string]*pssSub)
// Set some extra data
sn.EnodeID = sn.Node.Server().NodeInfo().Enode
sn.PssPubKey = hexutil.Encode(crypto.FromECDSAPub(sn.Pss.PublicKey()))
sn.PssAddr, err = sn.Pss.BaseAddr()
if err != nil {
return fmt.Errorf("pss API fail %v", err)
}
// Print some information
log.Info(fmt.Sprintf("My PSS pubkey is %s", sn.PssPubKey))
log.Info(fmt.Sprintf("My PSS address is %x", sn.PssAddr))
// Run statistics goroutine
sn.PrintStats()
return nil
}
func strTopic(topic string) pss.Topic {
return pss.BytesToTopic([]byte(topic))
}
func strSymKey(key string) []byte {
symKey := make([]byte, 32)
copy(symKey, []byte(key))
return symKey
}
func strAddress(addr string) pss.PssAddress {
var pssAddress pss.PssAddress
pssAddress = []byte(addr)
return pssAddress
}
func (sn *SwarmNet) PssSub(subType, key, topic, address string) error {
pssTopic := strTopic(topic)
pssAddress := strAddress(address)
switch subType {
case "sym":
_, err := sn.Pss.SetSymmetricKey(strSymKey(key), pssTopic, pssAddress, true)
if err != nil {
return err
}
}
sn.PssTopics[topic] = new(pssSub)
sn.PssTopics[topic].Address = address
sn.PssTopics[topic].Delivery = make(chan []byte)
var pssHandler pss.HandlerFunc = func(msg []byte, peer *p2p.Peer, asym bool, keyid string) error {
//log.Info("pss received", "msg", fmt.Sprintf("%s", msg), "keyid", fmt.Sprintf("%s", keyid))
sn.PssTopics[topic].Delivery <- msg
return nil
}
topicHandler := pss.NewHandler(pssHandler)
sn.PssTopics[topic].Unregister = sn.Pss.Register(&pssTopic, topicHandler)
log.Info(fmt.Sprintf("Subscribed to topic %s", pssTopic.String()))
return nil
}
func (sn *SwarmNet) PssPub(subType, key, topic, msg, address string) error {
var err error
dstAddr := strAddress(address)
dstTopic := strTopic(topic)
if subType == "sym" {
symKeyId, err := sn.Pss.SetSymmetricKey(strSymKey(key), dstTopic, dstAddr, false)
if err != nil {
return err
}
err = sn.Pss.SendSym(symKeyId, strTopic(topic), hexutil.Bytes(msg))
}
if subType == "raw" {
err = sn.Pss.SendRaw(hexutil.Bytes(dstAddr), dstTopic, hexutil.Bytes(msg))
}
if subType == "asym" {
if hasHexPrefix := strings.HasPrefix(key, "0x"); !hasHexPrefix {
key = "0x" + key
}
topics, addresses, err := sn.Pss.GetPublickeyPeers(key)
if err != nil {
return err
}
topicFound := false
for i, t := range topics {
if dstTopic == t && fmt.Sprintf("%x", addresses[i]) == fmt.Sprintf("%x", dstAddr) {
topicFound = true
break
}
}
if !topicFound {
pubKeyBytes, err := hexutil.Decode(key)
if err != nil {
return err
}
err = sn.Pss.SetPeerPublicKey(pubKeyBytes, dstTopic, dstAddr)
if err != nil {
return err
}
}
err = sn.Pss.SendAsym(key, dstTopic, hexutil.Bytes(msg))
}
return err
}
func (sn *SwarmNet) Test() error {
sn.PssSub("sym", "vocdoni", "vocdoni_test", "")
go func() {
for {
msg := <-sn.PssTopics["vocdoni_test"].Delivery
fmt.Printf("Pss received: %s\n", msg)
}
}()
hostname, _ := os.Hostname()
for {
err := sn.PssPub("sym", "vocdoni", "vocdoni_test", fmt.Sprintf("Hello world from %s", hostname), "")
log.Info("pss sent", "err", err)
time.Sleep(10 * time.Second)
}
}