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