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