swarm, cmd/swarm: Enable pss

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
 }
 

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 {

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,22 +65,23 @@ 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(),
+		HiveParams:    network.NewHiveParams(),
-		SyncParams:    network.NewDefaultSyncParams(),
+		//SyncParams:    network.NewDefaultSyncParams(),
 		Swap:        swap.NewDefaultSwapParams(),
 		ListenAddr:  DefaultHTTPListenAddr,
 		Port:        DefaultHTTPPort,
 		Path:        node.DefaultDataDir(),
 		EnsApi:      node.DefaultIPCEndpoint("geth"),
 		EnsRoot:     ens.TestNetAddress,
-		NetworkId:     network.NetworkId,
+		NetworkId:   network.NetworkID,
 		SwapEnabled: false,
 		SyncEnabled: true,
+		PssEnabled:  true,
 		SwapApi:     "",
 		BootNodes:   "",
 	}
@@ -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.SyncParams.Init(self.Path)
-	self.HiveParams.Init(self.Path)
+	//self.HiveParams.Init(self.Path)
 	self.StoreParams.Init(self.Path)
 }

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,11 +32,13 @@ 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"
 )
 
@@ -46,18 +47,19 @@ 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
+	//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
+	//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
+	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)
+	kp := network.NewKadParams()
-	log.Debug(fmt.Sprintf("Set up local db access (iterator/counter)"))
+	to := network.NewKademlia(
-
+		common.FromHex(config.BzzKey),
-	// set up the kademlia hive
+		kp,
-	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
 	)
-	log.Debug(fmt.Sprintf("Set up swarm network with Kademlia hive"))
 
-	// setup cloud storage backend
+	config.HiveParams.Discovery = true
-	self.cloud = network.NewForwarder(self.hive)
-	log.Debug(fmt.Sprintf("-> set swarm forwarder as cloud storage backend"))
 
 	// 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"))
-
+	nodeid := discover.PubkeyID(crypto.ToECDSAPub(common.FromHex(config.PublicKey)))
-	// set up Depo (storage handler = cloud storage access layer for incoming remote requests)
+	addr := network.NewAddrFromNodeID(nodeid)
-	self.depo = network.NewDepo(hash, self.lstore, self.storage)
+	bzzconfig := &network.BzzConfig{
-	log.Debug(fmt.Sprintf("-> REmote Access to CHunks"))
+		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 {
+func (self *Swarm) Start(net *p2p.Server) error {
-	connectPeer := func(url string) error {
+
-		node, err := discover.ParseNode(url)
+	// update uaddr to correct enode
-		if err != nil {
+	newaddr := self.bzz.UpdateLocalAddr([]byte(net.Self().String()))
-			return fmt.Errorf("invalid node URL: %v", err)
+	log.Warn("Updated bzz local addr", "oaddr", fmt.Sprintf("%x", newaddr.OAddr), "uaddr", fmt.Sprintf("%x", newaddr.UAddr))
-		}
+
-		srv.AddPeer(node)
-		return nil
-	}
 	// set chequebook
 	if self.swapEnabled {
 		ctx := context.Background() // The initial setup has no deadline.
@@ -189,29 +193,36 @@ func (self *Swarm) Start(srv *p2p.Server) error {
 	}
 
 	log.Warn(fmt.Sprintf("Starting Swarm service"))
-	self.hive.Start(
+
-		discover.PubkeyID(&srv.PrivateKey.PublicKey),
+	err := self.bzz.Start(net)
-		func() string { return srv.ListenAddr },
+	if err != nil {
-		connectPeer,
+		log.Error("bzz failed", "err", err)
-	)
+		return err
-	log.Info(fmt.Sprintf("Swarm network started on bzz address: %v", self.hive.Addr()))
+	}
+	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))
+	}
+
+	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 {
+func (self *Swarm) Protocols() (protos []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 {
+	for _, p := range self.bzz.Protocols() {
-		return nil
+		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 {