diff --git a/cmd/swarm/config.go b/cmd/swarm/config.go index 33235ca06e..8b1d87f142 100644 --- a/cmd/swarm/config.go +++ b/cmd/swarm/config.go @@ -69,6 +69,7 @@ const ( SWARM_ENV_ENS_ADDR = "SWARM_ENS_ADDR" SWARM_ENV_CORS = "SWARM_CORS" SWARM_ENV_BOOTNODES = "SWARM_BOOTNODES" + SWARM_ENV_PSS_ENABLE = "SWARM_PSS_ENABLE" GETH_ENV_DATADIR = "GETH_DATADIR" ) @@ -94,7 +95,7 @@ func buildConfig(ctx *cli.Context) (config *bzzapi.Config, err error) { //check for deprecated flags checkDeprecated(ctx) //start by creating a default config - config = bzzapi.NewDefaultConfig() + config = bzzapi.NewConfig() //first load settings from config file (if provided) config, err = configFileOverride(config, ctx) //override settings provided by environment variables @@ -211,6 +212,10 @@ func cmdLineOverride(currentConfig *bzzapi.Config, ctx *cli.Context) *bzzapi.Con currentConfig.BootNodes = ctx.GlobalString(utils.BootnodesFlag.Name) } + if ctx.GlobalIsSet(SwarmPssEnabledFlag.Name) { + currentConfig.PssEnabled = true + } + return currentConfig } @@ -283,6 +288,12 @@ func envVarsOverride(currentConfig *bzzapi.Config) (config *bzzapi.Config) { currentConfig.BootNodes = bootnodes } + if pssenable := os.Getenv(SWARM_ENV_PSS_ENABLE); pssenable != "" { + if ps, err := strconv.ParseBool(pssenable); err != nil { + currentConfig.PssEnabled = ps + } + } + return currentConfig } diff --git a/cmd/swarm/main.go b/cmd/swarm/main.go index 77315a4265..2b216a0036 100644 --- a/cmd/swarm/main.go +++ b/cmd/swarm/main.go @@ -145,6 +145,10 @@ var ( Name: "mime", Usage: "force mime type", } + SwarmPssEnabledFlag = cli.BoolFlag{ + Name: "pss", + Usage: "Enable pss (message passing over swarm)", + } CorsStringFlag = cli.StringFlag{ Name: "corsdomain", Usage: "Domain on which to send Access-Control-Allow-Origin header (multiple domains can be supplied separated by a ',')", @@ -361,9 +365,19 @@ DEPRECATED: use 'swarm db clean'. SwarmUploadDefaultPath, SwarmUpFromStdinFlag, SwarmUploadMimeType, + // pss flags + SwarmPssEnabledFlag, //deprecated flags DeprecatedEthAPIFlag, } + rpcFlags := []cli.Flag{ + utils.WSEnabledFlag, + utils.WSListenAddrFlag, + utils.WSPortFlag, + utils.WSApiFlag, + utils.WSAllowedOriginsFlag, + } + app.Flags = append(app.Flags, rpcFlags...) app.Flags = append(app.Flags, debug.Flags...) app.Before = func(ctx *cli.Context) error { runtime.GOMAXPROCS(runtime.NumCPU()) @@ -514,7 +528,7 @@ func registerBzzService(bzzconfig *bzzapi.Config, ctx *cli.Context, stack *node. } } - return swarm.NewSwarm(ctx, swapClient, ensClient, bzzconfig, bzzconfig.SwapEnabled, bzzconfig.SyncEnabled, bzzconfig.Cors) + return swarm.NewSwarm(ctx, swapClient, ensClient, bzzconfig, bzzconfig.SwapEnabled, bzzconfig.SyncEnabled, bzzconfig.Cors, bzzconfig.PssEnabled) } //register within the ethereum node if err := stack.Register(boot); err != nil { diff --git a/swarm/api/config.go b/swarm/api/config.go index 140c938ae0..d4dba36094 100644 --- a/swarm/api/config.go +++ b/swarm/api/config.go @@ -45,7 +45,7 @@ type Config struct { *storage.ChunkerParams *network.HiveParams Swap *swap.SwapParams - *network.SyncParams + //*network.SyncParams Contract common.Address EnsRoot common.Address EnsApi string @@ -57,6 +57,7 @@ type Config struct { NetworkId uint64 SwapEnabled bool SyncEnabled bool + PssEnabled bool SwapApi string Cors string BzzAccount string @@ -64,24 +65,25 @@ type Config struct { } //create a default config with all parameters to set to defaults -func NewDefaultConfig() (self *Config) { +func NewConfig() (self *Config) { self = &Config{ StoreParams: storage.NewDefaultStoreParams(), ChunkerParams: storage.NewChunkerParams(), - HiveParams: network.NewDefaultHiveParams(), - SyncParams: network.NewDefaultSyncParams(), - Swap: swap.NewDefaultSwapParams(), - ListenAddr: DefaultHTTPListenAddr, - Port: DefaultHTTPPort, - Path: node.DefaultDataDir(), - EnsApi: node.DefaultIPCEndpoint("geth"), - EnsRoot: ens.TestNetAddress, - NetworkId: network.NetworkId, - SwapEnabled: false, - SyncEnabled: true, - SwapApi: "", - BootNodes: "", + HiveParams: network.NewHiveParams(), + //SyncParams: network.NewDefaultSyncParams(), + Swap: swap.NewDefaultSwapParams(), + ListenAddr: DefaultHTTPListenAddr, + Port: DefaultHTTPPort, + Path: node.DefaultDataDir(), + EnsApi: node.DefaultIPCEndpoint("geth"), + EnsRoot: ens.TestNetAddress, + NetworkId: network.NetworkID, + SwapEnabled: false, + SyncEnabled: true, + PssEnabled: true, + SwapApi: "", + BootNodes: "", } return @@ -107,7 +109,7 @@ func (self *Config) Init(prvKey *ecdsa.PrivateKey) { self.BzzKey = keyhex self.Swap.Init(self.Contract, prvKey) - self.SyncParams.Init(self.Path) - self.HiveParams.Init(self.Path) + //self.SyncParams.Init(self.Path) + //self.HiveParams.Init(self.Path) self.StoreParams.Init(self.Path) } diff --git a/swarm/swarm.go b/swarm/swarm.go index 3be3660b58..8a3128fbc6 100644 --- a/swarm/swarm.go +++ b/swarm/swarm.go @@ -21,7 +21,6 @@ import ( "context" "crypto/ecdsa" "fmt" - "net" "github.com/ethereum/go-ethereum/accounts/abi/bind" "github.com/ethereum/go-ethereum/common" @@ -33,31 +32,34 @@ import ( "github.com/ethereum/go-ethereum/node" "github.com/ethereum/go-ethereum/p2p" "github.com/ethereum/go-ethereum/p2p/discover" + "github.com/ethereum/go-ethereum/p2p/protocols" "github.com/ethereum/go-ethereum/rpc" "github.com/ethereum/go-ethereum/swarm/api" httpapi "github.com/ethereum/go-ethereum/swarm/api/http" "github.com/ethereum/go-ethereum/swarm/fuse" "github.com/ethereum/go-ethereum/swarm/network" + "github.com/ethereum/go-ethereum/swarm/pss" "github.com/ethereum/go-ethereum/swarm/storage" ) // the swarm stack type Swarm struct { - config *api.Config // swarm configuration - api *api.Api // high level api layer (fs/manifest) - dns api.Resolver // DNS registrar - dbAccess *network.DbAccess // access to local chunk db iterator and storage counter - storage storage.ChunkStore // internal access to storage, common interface to cloud storage backends - dpa *storage.DPA // distributed preimage archive, the local API to the storage with document level storage/retrieval support - depo network.StorageHandler // remote request handler, interface between bzz protocol and the storage - cloud storage.CloudStore // procurement, cloud storage backend (can multi-cloud) - hive *network.Hive // the logistic manager - backend chequebook.Backend // simple blockchain Backend + config *api.Config // swarm configuration + api *api.Api // high level api layer (fs/manifest) + dns api.Resolver // DNS registrar + //dbAccess *network.DbAccess // access to local chunk db iterator and storage counter + storage storage.ChunkStore // internal access to storage, common interface to cloud storage backends + dpa *storage.DPA // distributed preimage archive, the local API to the storage with document level storage/retrieval support + //depo network.StorageHandler // remote request handler, interface between bzz protocol and the storage + cloud storage.CloudStore // procurement, cloud storage backend (can multi-cloud) + bzz *network.Bzz // the logistic manager + backend chequebook.Backend // simple blockchain Backend privateKey *ecdsa.PrivateKey corsString string swapEnabled bool lstore *storage.LocalStore // local store, needs to store for releasing resources after node stopped sfs *fuse.SwarmFS // need this to cleanup all the active mounts on node exit + ps *pss.Pss } type SwarmAPI struct { @@ -76,7 +78,7 @@ func (self *Swarm) API() *SwarmAPI { // creates a new swarm service instance // implements node.Service -func NewSwarm(ctx *node.ServiceContext, backend chequebook.Backend, ensClient *ethclient.Client, config *api.Config, swapEnabled, syncEnabled bool, cors string) (self *Swarm, err error) { +func NewSwarm(ctx *node.ServiceContext, backend chequebook.Backend, ensClient *ethclient.Client, config *api.Config, swapEnabled, syncEnabled bool, cors string, pssEnabled bool) (self *Swarm, err error) { if bytes.Equal(common.FromHex(config.PublicKey), storage.ZeroKey) { return nil, fmt.Errorf("empty public key") } @@ -102,29 +104,25 @@ func NewSwarm(ctx *node.ServiceContext, backend chequebook.Backend, ensClient *e // setup local store log.Debug(fmt.Sprintf("Set up local storage")) - self.dbAccess = network.NewDbAccess(self.lstore) - log.Debug(fmt.Sprintf("Set up local db access (iterator/counter)")) - - // set up the kademlia hive - self.hive = network.NewHive( - common.HexToHash(self.config.BzzKey), // key to hive (kademlia base address) - config.HiveParams, // configuration parameters - swapEnabled, // SWAP enabled - syncEnabled, // syncronisation enabled + kp := network.NewKadParams() + to := network.NewKademlia( + common.FromHex(config.BzzKey), + kp, ) - log.Debug(fmt.Sprintf("Set up swarm network with Kademlia hive")) - // setup cloud storage backend - self.cloud = network.NewForwarder(self.hive) - log.Debug(fmt.Sprintf("-> set swarm forwarder as cloud storage backend")) + config.HiveParams.Discovery = true // setup cloud storage internal access layer self.storage = storage.NewNetStore(hash, self.lstore, self.cloud, config.StoreParams) log.Debug(fmt.Sprintf("-> swarm net store shared access layer to Swarm Chunk Store")) - - // set up Depo (storage handler = cloud storage access layer for incoming remote requests) - self.depo = network.NewDepo(hash, self.lstore, self.storage) - log.Debug(fmt.Sprintf("-> REmote Access to CHunks")) + nodeid := discover.PubkeyID(crypto.ToECDSAPub(common.FromHex(config.PublicKey))) + addr := network.NewAddrFromNodeID(nodeid) + bzzconfig := &network.BzzConfig{ + OverlayAddr: common.FromHex(config.BzzKey), + UnderlayAddr: addr.UAddr, + HiveParams: config.HiveParams, + } + self.bzz = network.NewBzz(bzzconfig, to, nil) // set up DPA, the cloud storage local access layer dpaChunkStore := storage.NewDpaChunkStore(self.lstore, self.storage) @@ -133,6 +131,15 @@ func NewSwarm(ctx *node.ServiceContext, backend chequebook.Backend, ensClient *e self.dpa = storage.NewDPA(dpaChunkStore, self.config.ChunkerParams) log.Debug(fmt.Sprintf("-> Content Store API")) + // Pss = postal service over swarm (devp2p over bzz) + if pssEnabled { + pssparams := pss.NewPssParams(self.privateKey) + self.ps = pss.NewPss(to, self.dpa, pssparams) + if pss.IsActiveHandshake { + pss.SetHandshakeController(self.ps, pss.NewHandshakeParams()) + } + } + // set up high level api transactOpts := bind.NewKeyedTransactor(self.privateKey) @@ -167,15 +174,12 @@ Start is called when the stack is started * TODO: start subservices like sword, swear, swarmdns */ // implements the node.Service interface -func (self *Swarm) Start(srv *p2p.Server) error { - connectPeer := func(url string) error { - node, err := discover.ParseNode(url) - if err != nil { - return fmt.Errorf("invalid node URL: %v", err) - } - srv.AddPeer(node) - return nil - } +func (self *Swarm) Start(net *p2p.Server) error { + + // update uaddr to correct enode + newaddr := self.bzz.UpdateLocalAddr([]byte(net.Self().String())) + log.Warn("Updated bzz local addr", "oaddr", fmt.Sprintf("%x", newaddr.OAddr), "uaddr", fmt.Sprintf("%x", newaddr.UAddr)) + // set chequebook if self.swapEnabled { ctx := context.Background() // The initial setup has no deadline. @@ -189,28 +193,35 @@ func (self *Swarm) Start(srv *p2p.Server) error { } log.Warn(fmt.Sprintf("Starting Swarm service")) - self.hive.Start( - discover.PubkeyID(&srv.PrivateKey.PublicKey), - func() string { return srv.ListenAddr }, - connectPeer, - ) - log.Info(fmt.Sprintf("Swarm network started on bzz address: %v", self.hive.Addr())) + + err := self.bzz.Start(net) + if err != nil { + log.Error("bzz failed", "err", err) + return err + } + log.Info(fmt.Sprintf("Swarm network started on bzz address: %x", self.bzz.Hive.Overlay.BaseAddr())) + + if self.ps != nil { + self.ps.Start(net) + log.Info("Pss started") + } self.dpa.Start() log.Debug(fmt.Sprintf("Swarm DPA started")) // start swarm http proxy server if self.config.Port != "" { - addr := net.JoinHostPort(self.config.ListenAddr, self.config.Port) + addr := ":" + self.config.Port go httpapi.StartHttpServer(self.api, &httpapi.ServerConfig{ Addr: addr, CorsString: self.corsString, }) - log.Info(fmt.Sprintf("Swarm http proxy started on %v", addr)) + } - if self.corsString != "" { - log.Debug(fmt.Sprintf("Swarm http proxy started with corsdomain: %v", self.corsString)) - } + log.Debug(fmt.Sprintf("Swarm http proxy started on port: %v", self.config.Port)) + + if self.corsString != "" { + log.Debug(fmt.Sprintf("Swarm http proxy started with corsdomain: %v", self.corsString)) } return nil @@ -220,7 +231,10 @@ func (self *Swarm) Start(srv *p2p.Server) error { // stops all component services. func (self *Swarm) Stop() error { self.dpa.Stop() - err := self.hive.Stop() + self.bzz.Stop() + if self.ps != nil { + self.ps.Stop() + } if ch := self.config.Swap.Chequebook(); ch != nil { ch.Stop() ch.Save() @@ -230,22 +244,38 @@ func (self *Swarm) Stop() error { self.lstore.DbStore.Close() } self.sfs.Stop() - return err + return nil } // implements the node.Service interface -func (self *Swarm) Protocols() []p2p.Protocol { - proto, err := network.Bzz(self.depo, self.backend, self.hive, self.dbAccess, self.config.Swap, self.config.SyncParams, self.config.NetworkId) - if err != nil { - return nil +func (self *Swarm) Protocols() (protos []p2p.Protocol) { + + for _, p := range self.bzz.Protocols() { + protos = append(protos, p) } - return []p2p.Protocol{proto} + + if self.ps != nil { + log.Warn("adding pss protos") + for _, p := range self.ps.Protocols() { + protos = append(protos, p) + } + } + return +} + +func (self *Swarm) RegisterPssProtocol(spec *protocols.Spec, targetprotocol *p2p.Protocol, options *pss.ProtocolParams) (*pss.Protocol, error) { + if !pss.IsActiveProtocol { + return nil, fmt.Errorf("Pss protocols not available (built with !nopssprotocol tag)") + } + topic := pss.ProtocolTopic(spec) + return pss.RegisterProtocol(self.ps, &topic, spec, targetprotocol, options) } // implements node.Service // Apis returns the RPC Api descriptors the Swarm implementation offers func (self *Swarm) APIs() []rpc.API { - return []rpc.API{ + + apis := []rpc.API{ // public APIs { Namespace: "bzz", @@ -257,7 +287,7 @@ func (self *Swarm) APIs() []rpc.API { { Namespace: "bzz", Version: "0.1", - Service: api.NewControl(self.api, self.hive), + Service: api.NewControl(self.api, self.bzz.Hive), Public: false, }, { @@ -288,6 +318,18 @@ func (self *Swarm) APIs() []rpc.API { }, // {Namespace, Version, api.NewAdmin(self), false}, } + + for _, api := range self.bzz.APIs() { + apis = append(apis, api) + } + + if self.ps != nil { + for _, api := range self.ps.APIs() { + apis = append(apis, api) + } + } + + return apis } func (self *Swarm) Api() *api.Api { @@ -301,7 +343,6 @@ func (self *Swarm) SetChequebook(ctx context.Context) error { return err } log.Info(fmt.Sprintf("new chequebook set (%v): saving config file, resetting all connections in the hive", self.config.Swap.Contract.Hex())) - self.hive.DropAll() return nil } @@ -313,10 +354,10 @@ func NewLocalSwarm(datadir, port string) (self *Swarm, err error) { return } - config := api.NewDefaultConfig() + config := api.NewConfig() config.Path = datadir - config.Init(prvKey) config.Port = port + config.Init(prvKey) dpa, err := storage.NewLocalDPA(datadir) if err != nil {