go-dvote/net/swarm/swarm.go
net/swarm/swarm.goRaw
1package swarm
2
3import (
4 "context"
5 "crypto/ecdsa"
6 "fmt"
7 "os"
8 "os/user"
9 "time"
10
11 "github.com/ethereum/go-ethereum/common/hexutil"
12 "github.com/ethereum/go-ethereum/node"
13
14 "github.com/ethereum/go-ethereum/crypto"
15 "github.com/ethereum/go-ethereum/p2p"
16
17 "github.com/ethereum/go-ethereum/log"
18 "github.com/ethereum/go-ethereum/p2p/enode"
19 "github.com/ethereum/go-ethereum/swarm"
20
21 swarmapi "github.com/ethereum/go-ethereum/swarm/api"
22 "github.com/ethereum/go-ethereum/swarm/network"
23 "github.com/ethereum/go-ethereum/swarm/pss"
24)
25
26const (
27 // MaxPeers is the maximum number of p2p peer connections
28 MaxPeers = 10
29)
30
31// SwarmBootnodes list of bootnodes for the SWARM network
32var SwarmBootnodes = []string{
33 // EF Swarm Bootnode - AWS - eu-central-1
34 "enode://4c113504601930bf2000c29bcd98d1716b6167749f58bad703bae338332fe93cc9d9204f08afb44100dc7bea479205f5d162df579f9a8f76f8b402d339709023@3.122.203.99:30301",
35 // EF Swarm Bootnode - AWS - us-west-2
36 "enode://89f2ede3371bff1ad9f2088f2012984e280287a4e2b68007c2a6ad994909c51886b4a8e9e2ecc97f9910aca538398e0a5804b0ee80a187fde1ba4f32626322ba@52.35.212.179:30301",
37}
38
39func newNode(key *ecdsa.PrivateKey, port int, httpport int, wsport int,
40 datadir string, modules ...string) (*node.Node, *node.Config, error) {
41 if port == 0 {
42 port = 30100
43 }
44 cfg := &node.DefaultConfig
45 if key != nil {
46 cfg.P2P.PrivateKey = key
47 }
48 cfg.P2P.MaxPeers = MaxPeers
49 cfg.P2P.ListenAddr = fmt.Sprintf("0.0.0.0:%d", port)
50 cfg.P2P.EnableMsgEvents = true
51 cfg.P2P.NoDiscovery = false
52 cfg.P2P.DiscoveryV5 = true
53 cfg.IPCPath = datadir + "/node.ipc"
54 cfg.DataDir = datadir
55 if httpport > 0 {
56 cfg.HTTPHost = node.DefaultHTTPHost
57 cfg.HTTPPort = httpport
58 cfg.HTTPCors = []string{"*"}
59 }
60 if wsport > 0 {
61 cfg.WSHost = node.DefaultWSHost
62 cfg.WSPort = wsport
63 cfg.WSOrigins = []string{"*"}
64 for i := 0; i < len(modules); i++ {
65 cfg.WSModules = append(cfg.WSModules, modules[i])
66 }
67 }
68 stack, err := node.New(cfg)
69 if err != nil {
70 return nil, nil, fmt.Errorf("ServiceNode create fail: %v", err)
71 }
72 return stack, cfg, nil
73}
74
75func newSwarm(privkey *ecdsa.PrivateKey, datadir string, port int) (*swarm.Swarm, *swarmapi.Config, node.ServiceConstructor) {
76 // create swarm service
77 swarmCfg := swarmapi.NewConfig()
78 swarmCfg.SyncEnabled = true
79 swarmCfg.Port = fmt.Sprintf("%d", port)
80 swarmCfg.Path = datadir
81 swarmCfg.HiveParams.Discovery = true
82 swarmCfg.Discovery = true
83 swarmCfg.Pss.MsgTTL = time.Second * 10
84 swarmCfg.Pss.CacheTTL = time.Second * 30
85 swarmCfg.Pss.AllowRaw = true
86 swarmCfg.Init(privkey)
87 swarmNode, err := swarm.NewSwarm(swarmCfg, nil)
88 if err != nil {
89 log.Crit("cannot crate swarm node")
90 }
91 // register swarm service to the node
92 var swarmService node.ServiceConstructor = func(ctx *node.ServiceContext) (node.Service, error) {
93 return swarmNode, nil
94 }
95 return swarmNode, swarmCfg, swarmService
96}
97
98type swarmPorts struct {
99 WebSockets int
100 HTTPRPC int
101 Bzz int
102 P2P int
103}
104
105func NewSwarmPorts() *swarmPorts {
106 sp := new(swarmPorts)
107 sp.WebSockets = 8544
108 sp.HTTPRPC = 8543
109 sp.Bzz = 8542
110 sp.P2P = 31000
111 return sp
112}
113
114type pssSub struct {
115 Unregister func()
116 Delivery (chan []byte)
117 Address string
118}
119
120type SwarmNet struct {
121 Node *node.Node
122 NodeConfig *node.Config
123 EnodeID string
124 Datadir string
125 Key *ecdsa.PrivateKey
126 Pss *pss.API
127 PssPubKey string
128 PssAddr pss.PssAddress
129 PssTopics map[string]*pssSub
130 Hive *network.Hive
131 Ports *swarmPorts
132}
133
134func (sn *SwarmNet) SetLog() {
135 // ensure good log formats for terminal
136 // handle verbosity flag
137 hs := log.StreamHandler(os.Stderr, log.TerminalFormat(true))
138 loglevel := log.LvlInfo
139 hf := log.LvlFilterHandler(loglevel, hs)
140 h := log.CallerFileHandler(hf)
141 log.Root().SetHandler(h)
142}
143
144func (sn *SwarmNet) PrintStats() {
145 // statistics thread
146 go func() {
147 for {
148 if sn.Node.Server() != nil && sn.Hive != nil {
149 addr := fmt.Sprintf("%x", sn.PssAddr)
150 var addrs [][]byte
151 addrs = append(addrs, []byte(addr))
152 peerCount := sn.Node.Server().PeerCount()
153 log.Info(fmt.Sprintf("PeerCount:%d NeighDepth:%d", peerCount, sn.Hive.NeighbourhoodDepth))
154 }
155 time.Sleep(time.Second * 5)
156 }
157 }()
158}
159
160func (sn *SwarmNet) SetDatadir(datadir string) {
161 sn.Datadir = datadir
162}
163
164func (sn *SwarmNet) SetKey(key *ecdsa.PrivateKey) {
165 sn.Key = key
166}
167
168func (sn *SwarmNet) Init() error {
169 var err error
170 if len(sn.Datadir) < 1 {
171 usr, err := user.Current()
172 if err != nil {
173 return err
174 }
175 sn.Datadir = usr.HomeDir + "/.dvote/swarm"
176 os.MkdirAll(sn.Datadir, 0755)
177 }
178
179 sn.SetLog()
180 sn.Ports = NewSwarmPorts()
181
182 // create node
183 sn.Node, sn.NodeConfig, err = newNode(sn.Key, sn.Ports.P2P,
184 sn.Ports.HTTPRPC, sn.Ports.WebSockets, sn.Datadir, "pss")
185 if err != nil {
186 return err
187 }
188 // set node key, if not set use the storage one or generate it
189 if sn.Key == nil {
190 sn.Key = sn.NodeConfig.NodeKey()
191 }
192
193 // create and register Swarm service
194 swarmNode, _, swarmHandler := newSwarm(sn.Key, sn.Datadir, sn.Ports.Bzz)
195 err = sn.Node.Register(swarmHandler)
196 if err != nil {
197 return fmt.Errorf("swarm register fail %v", err)
198 }
199
200 // start the node
201 sn.Node.Start()
202 for _, url := range SwarmBootnodes {
203 log.Info("Add bootnode " + url)
204 node, _ := enode.ParseV4(url)
205 sn.Node.Server().AddPeer(node)
206 }
207
208 // wait to connect to the p2p network
209 _, cancel := context.WithTimeout(context.Background(), time.Second)
210 defer cancel()
211 time.Sleep(time.Second * 5)
212
213 // Get the services API
214 for _, a := range swarmNode.APIs() {
215 switch a.Service.(type) {
216 case *network.Hive:
217 sn.Hive = a.Service.(*network.Hive)
218 case *pss.API:
219 sn.Pss = a.Service.(*pss.API)
220 }
221 }
222
223 // Create topics map
224 sn.PssTopics = make(map[string]*pssSub)
225
226 // Set some extra data
227 sn.EnodeID = sn.Node.Server().NodeInfo().Enode
228 sn.PssPubKey = hexutil.Encode(crypto.FromECDSAPub(sn.Pss.PublicKey()))
229 sn.PssAddr, err = sn.Pss.BaseAddr()
230 if err != nil {
231 return fmt.Errorf("pss API fail %v", err)
232 }
233
234 // Print some information
235 log.Info(fmt.Sprintf("My PSS pubkey is %s", sn.PssPubKey))
236 log.Info(fmt.Sprintf("My PSS address is %x", sn.PssAddr))
237
238 // Run statistics goroutine
239 sn.PrintStats()
240
241 return nil
242}
243
244func strTopic(topic string) pss.Topic {
245 return pss.BytesToTopic([]byte(topic))
246}
247
248func strSymKey(key string) []byte {
249 symKey := make([]byte, 32)
250 copy(symKey, []byte(key))
251 return symKey
252}
253
254func strAddress(addr string) pss.PssAddress {
255 var pssAddress pss.PssAddress
256 pssAddress = []byte(addr)
257 return pssAddress
258}
259
260func (sn *SwarmNet) PssSub(subType, key, topic, address string) error {
261 pssTopic := strTopic(topic)
262 pssAddress := strAddress(address)
263 switch subType {
264 case "sym":
265 _, err := sn.Pss.SetSymmetricKey(strSymKey(key), pssTopic, pssAddress, true)
266 if err != nil {
267 return err
268 }
269 }
270
271 sn.PssTopics[topic] = new(pssSub)
272 sn.PssTopics[topic].Address = address
273 sn.PssTopics[topic].Delivery = make(chan []byte)
274
275 var pssHandler pss.HandlerFunc = func(msg []byte, peer *p2p.Peer, asym bool, keyid string) error {
276 //log.Info("pss received", "msg", fmt.Sprintf("%s", msg), "keyid", fmt.Sprintf("%s", keyid))
277 sn.PssTopics[topic].Delivery <- msg
278 return nil
279 }
280 topicHandler := pss.NewHandler(pssHandler)
281 sn.PssTopics[topic].Unregister = sn.Pss.Register(&pssTopic, topicHandler)
282
283 log.Info(fmt.Sprintf("Subscribed to topic %s", pssTopic.String()))
284 return nil
285}
286
287func (sn *SwarmNet) PssPub(subType, key, topic, msg, address string) error {
288 var err error
289 dstAddr := strAddress(address)
290 dstTopic := strTopic(topic)
291 if subType == "sym" {
292 symKeyId, err := sn.Pss.SetSymmetricKey(strSymKey(key), dstTopic, dstAddr, false)
293 if err != nil {
294 return err
295 }
296 err = sn.Pss.SendSym(symKeyId, strTopic(topic), hexutil.Bytes(msg))
297 }
298 if subType == "raw" {
299 err = sn.Pss.SendRaw(hexutil.Bytes(dstAddr), dstTopic, hexutil.Bytes(msg))
300 }
301 if subType == "asym" {
302 err = sn.Pss.SetPeerPublicKey(hexutil.Bytes(key), dstTopic, dstAddr)
303 if err != nil {
304 return err
305 }
306 err = sn.Pss.SendAsym(key, dstTopic, hexutil.Bytes(msg))
307 }
308 return err
309}
310
311func (sn *SwarmNet) Test() error {
312 sn.PssSub("sym", "vocdoni", "vocdoni_test", "")
313
314 go func() {
315 for {
316 msg := <-sn.PssTopics["vocdoni_test"].Delivery
317 fmt.Printf("Pss received: %s\n", msg)
318 }
319 }()
320
321 hostname, _ := os.Hostname()
322 for {
323 err := sn.PssPub("sym", "vocdoni", "vocdoni_test", fmt.Sprintf("Hello world from %s", hostname), "")
324 log.Info("pss sent", "err", err)
325 time.Sleep(10 * time.Second)
326 }
327}