swarm, swarm/pss, whisper: ssh topics, asym encrypt

swarm/pss/ping.go

@@ -1,13 +1,16 @@
 package pss
 
 import (
+	"crypto/ecdsa"
 	"fmt"
 	"io/ioutil"
 	"os"
 	"time"
 
+	"github.com/ethereum/go-ethereum/crypto"
 	"github.com/ethereum/go-ethereum/log"
 	"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/swarm/network"
 	"github.com/ethereum/go-ethereum/swarm/storage"
@@ -39,28 +42,6 @@ var PingProtocol = &protocols.Spec{
 
 var PingTopic = NewTopic(PingProtocol.Name, int(PingProtocol.Version))
 
-func NewPingMsg(to []byte, spec *protocols.Spec, topic Topic, senderaddr []byte) PssMsg {
-	data := PingMsg{
-		Created: time.Now(),
-	}
-	code, found := spec.GetCode(&data)
-	if !found {
-		return PssMsg{}
-	}
-
-	rlpbundle, err := NewProtocolMsg(code, data)
-	if err != nil {
-		return PssMsg{}
-	}
-
-	pssmsg := PssMsg{
-		To:      to,
-		Payload: NewEnvelope(senderaddr, topic, rlpbundle),
-	}
-
-	return pssmsg
-}
-
 func NewPingProtocol(handler func(interface{}) error) *p2p.Protocol {
 	return &p2p.Protocol{
 		Name:    PingProtocol.Name,
@@ -75,10 +56,11 @@ func NewPingProtocol(handler func(interface{}) error) *p2p.Protocol {
 	}
 }
 
-func NewTestPss(addr []byte) *Pss {
+func NewTestPss(privkey *ecdsa.PrivateKey) *Pss {
-	if addr == nil {
+
-		addr = network.RandomAddr().OAddr
+	var nid discover.NodeID
-	}
+	copy(nid[:], crypto.FromECDSAPub(&privkey.PublicKey))
+	addr := network.NewAddrFromNodeID(nid)
 
 	// set up storage
 	cachedir, err := ioutil.TempDir("", "pss-cache")
@@ -97,9 +79,9 @@ func NewTestPss(addr []byte) *Pss {
 	kp.MinProxBinSize = 3
 
 	// create pss
-	pp := NewPssParams(true)
+	pp := NewPssParams(privkey)
 
-	overlay := network.NewKademlia(addr, kp)
+	overlay := network.NewKademlia(addr.Over(), kp)
 	ps := NewPss(overlay, dpa, pp)
 
 	return ps

swarm/pss/pss.go

@@ -2,12 +2,14 @@ package pss
 
 import (
 	"bytes"
+	"crypto/ecdsa"
 	"errors"
 	"fmt"
 	"sync"
 	"time"
 
 	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/crypto"
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/p2p"
 	"github.com/ethereum/go-ethereum/p2p/discover"
@@ -17,6 +19,7 @@ import (
 	"github.com/ethereum/go-ethereum/rpc"
 	"github.com/ethereum/go-ethereum/swarm/network"
 	"github.com/ethereum/go-ethereum/swarm/storage"
+	whisper "github.com/ethereum/go-ethereum/whisper/whisperv5"
 )
 
 const (
@@ -28,6 +31,7 @@ const (
 
 var (
 	errorForwardToSelf = errors.New("forward to self")
+	errorWhisper       = errors.New("whisper backend")
 )
 
 // abstraction to enable access to p2p.protocols.Peer.Send
@@ -58,15 +62,17 @@ type pssDigest [digestLength]byte
 //
 // Implements node.Service
 type Pss struct {
-	network.Overlay                                             // we can get the overlayaddress from this
+	network.Overlay                                                         // we can get the overlayaddress from this
-	peerPool        map[pot.Address]map[Topic]p2p.MsgReadWriter // keep track of all virtual p2p.Peers we are currently speaking to
+	peerPool        map[pot.Address]map[whisper.TopicType]p2p.MsgReadWriter // keep track of all virtual p2p.Peers we are currently speaking to
-	fwdPool         map[discover.NodeID]*protocols.Peer         // keep track of all peers sitting on the pssmsg routing layer
+	fwdPool         map[discover.NodeID]*protocols.Peer                     // keep track of all peers sitting on the pssmsg routing layer
-	handlers        map[Topic]map[*Handler]bool                 // topic and version based pss payload handlers
+	keyPool         map[pot.Address]ecdsa.PublicKey                         // keep track of all public keys so we can encrypt for our peers
-	fwdcache        map[pssDigest]pssCacheEntry                 // checksum of unique fields from pssmsg mapped to expiry, cache to determine whether to drop msg
+	reverseKeyPool  map[string]pot.Address                                  // as above but reverse lookup
-	cachettl        time.Duration                               // how long to keep messages in fwdcache
+	handlers        map[whisper.TopicType]map[*Handler]bool                 // topic and version based pss payload handlers
+	fwdcache        map[pssDigest]pssCacheEntry                             // checksum of unique fields from pssmsg mapped to expiry, cache to determine whether to drop msg
+	cachettl        time.Duration                                           // how long to keep messages in fwdcache
 	lock            sync.Mutex
 	dpa             *storage.DPA
-	debug           bool
+	privatekey      *ecdsa.PrivateKey
 }
 
 func (self *Pss) storeMsg(msg *PssMsg) (pssDigest, error) {
@@ -89,14 +95,16 @@ func (self *Pss) storeMsg(msg *PssMsg) (pssDigest, error) {
 // Needs a swarm network overlay, a DPA storage for message cache storage.
 func NewPss(k network.Overlay, dpa *storage.DPA, params *PssParams) *Pss {
 	return &Pss{
-		Overlay:  k,
+		Overlay:        k,
-		peerPool: make(map[pot.Address]map[Topic]p2p.MsgReadWriter, PssPeerCapacity),
+		peerPool:       make(map[pot.Address]map[whisper.TopicType]p2p.MsgReadWriter, PssPeerCapacity),
-		fwdPool:  make(map[discover.NodeID]*protocols.Peer),
+		fwdPool:        make(map[discover.NodeID]*protocols.Peer),
-		handlers: make(map[Topic]map[*Handler]bool),
+		keyPool:        make(map[pot.Address]ecdsa.PublicKey),
-		fwdcache: make(map[pssDigest]pssCacheEntry),
+		reverseKeyPool: make(map[string]pot.Address),
-		cachettl: params.Cachettl,
+		handlers:       make(map[whisper.TopicType]map[*Handler]bool),
-		dpa:      dpa,
+		fwdcache:       make(map[pssDigest]pssCacheEntry),
-		debug:    params.Debug,
+		cachettl:       params.Cachettl,
+		dpa:            dpa,
+		privatekey:     params.privatekey,
 	}
 }
 
@@ -132,10 +140,7 @@ func (self *Pss) Protocols() []p2p.Protocol {
 // Starts the PssMsg protocol
 func (self *Pss) Run(p *p2p.Peer, rw p2p.MsgReadWriter) error {
 	pp := protocols.NewPeer(p, rw, pssSpec)
-	//addr := network.NewAddrFromNodeID(id)
-	//potaddr := pot.NewHashAddressFromBytes(addr.OAddr)
 	self.fwdPool[p.ID()] = pp
-
 	return pp.Run(self.handlePssMsg)
 }
 
@@ -151,14 +156,6 @@ func (self *Pss) APIs() []rpc.API {
 			Public:    true,
 		},
 	}
-	if self.debug {
-		apis = append(apis, rpc.API{
-			Namespace: "pss",
-			Version:   "0.1",
-			Service:   NewAPITest(self),
-			Public:    true,
-		})
-	}
 	return apis
 }
 
@@ -167,7 +164,7 @@ func (self *Pss) APIs() []rpc.API {
 // After calling this, all incoming messages with an envelope Topic matching the Topic specified will be passed to the given Handler function.
 //
 // Returns a deregister function which needs to be called to deregister the handler,
-func (self *Pss) Register(topic *Topic, handler Handler) func() {
+func (self *Pss) Register(topic *whisper.TopicType, handler Handler) func() {
 	self.lock.Lock()
 	defer self.lock.Unlock()
 	handlers := self.handlers[*topic]
@@ -179,7 +176,27 @@ func (self *Pss) Register(topic *Topic, handler Handler) func() {
 	return func() { self.deregister(topic, &handler) }
 }
 
-func (self *Pss) deregister(topic *Topic, h *Handler) {
+func (self *Pss) AddAddressKeyPair(addr pot.Address, pubkey ecdsa.PublicKey) {
+	self.lock.Lock()
+	defer self.lock.Unlock()
+	self.keyPool[addr] = pubkey
+	self.reverseKeyPool[common.ToHex(crypto.FromECDSAPub(&pubkey))] = addr
+}
+
+// may need these later, please let them be
+////func (self *Pss) RemoveAddressKeyPair(addr pot.Address, pubkey ecdsa.PublicKey) {
+//	delete(self.reverseKeyPool, &self.keyPool[addr])
+//	delete(self.keyPool, addr)
+//}
+//
+//func (self *Pss) hasAddressKeyPair(addr pot.Address, pubkey ecdsa.PublicKey) bool {
+//	if self.keyPool[addr] != nil {
+//		return *self.keyPool[addr] == pubkey
+//	}
+//	return false
+//}
+
+func (self *Pss) deregister(topic *whisper.TopicType, h *Handler) {
 	self.lock.Lock()
 	defer self.lock.Unlock()
 	handlers := self.handlers[*topic]
@@ -241,35 +258,56 @@ func (self *Pss) checkFwdCache(addr []byte, digest pssDigest) bool {
 	return false
 }
 
-func (self *Pss) getHandlers(topic Topic) map[*Handler]bool {
+func (self *Pss) getHandlers(topic whisper.TopicType) map[*Handler]bool {
 	self.lock.Lock()
 	defer self.lock.Unlock()
 	return self.handlers[topic]
 }
 
 func (self *Pss) handlePssMsg(msg interface{}) error {
-	pssmsg := msg.(*PssMsg)
+	pssmsg, ok := msg.(*PssMsg)
+	if ok {
+		if !self.isSelfRecipient(pssmsg) {
+			log.Trace("pss was for someone else :'( ... forwarding")
+			return self.Forward(pssmsg)
+		}
+		log.Trace("pss for us, yay! ... let's process!")
 
-	if !self.isSelfRecipient(pssmsg) {
+		return self.Process(pssmsg)
-		log.Trace("pss was for someone else :'( ... forwarding")
-		return self.Forward(pssmsg)
 	}
-	log.Trace("pss for us, yay! ... let's process!")
+
-	return self.Process(pssmsg)
+	return fmt.Errorf("invalid message")
 }
 
 // Entry point to processing a message for which the current node is the intended recipient.
 func (self *Pss) Process(pssmsg *PssMsg) error {
 	env := pssmsg.Payload
-	payload := env.Payload
+	recvmsg, err := env.OpenAsymmetric(self.privatekey)
+	if err != nil {
+		// todo: add check on if key is full length and identical, then fail
+		//return self.Forward(pssmsg)
+		return fmt.Errorf("not for us", "err", err)
+	}
+	if !recvmsg.Validate() {
+		return fmt.Errorf("invalid signature")
+	}
+
+	payload := recvmsg.Payload
 	handlers := self.getHandlers(env.Topic)
 	if len(handlers) == 0 {
 		return fmt.Errorf("No registered handler for topic '%x'", env.Topic)
 	}
+
 	nid, _ := discover.HexID("0x00")
-	p := p2p.NewPeer(nid, fmt.Sprintf("%x", env.From), []p2p.Cap{})
+	p := p2p.NewPeer(nid, fmt.Sprintf("%x", recvmsg.Src), []p2p.Cap{})
+	addr := self.reverseKeyPool[common.ToHex(crypto.FromECDSAPub(recvmsg.Src))]
+	log.Warn("recvkey", "key", *recvmsg.Src, "addr", addr)
+	if bytes.Equal([]byte{}, addr[:]) {
+		return fmt.Errorf("unknown key", "addr", addr)
+	}
+
 	for f := range handlers {
-		err := (*f)(payload, p, env.From)
+		err := (*f)(payload, p, addr.Bytes())
 		if err != nil {
 			return err
 		}
@@ -282,12 +320,38 @@ func (self *Pss) Process(pssmsg *PssMsg) error {
 // This method is payload agnostic, and will accept any arbitrary byte slice as the payload for a message.
 //
 // It generates an envelope for the specified recipient and topic, and wraps the message payload in it.
-func (self *Pss) SendRaw(to []byte, topic Topic, msg []byte) error {
+func (self *Pss) SendAsym(to []byte, topic whisper.TopicType, msg []byte) error {
-	sender := self.Overlay.BaseAddr()
+	var potaddr pot.Address
-	pssenv := NewEnvelope(sender, topic, msg)
+	copy(potaddr[:], to)
+	topubkey := self.keyPool[potaddr]
+	log.Debug("using pubkey", "pubkey", topubkey)
+	wparams := &whisper.MessageParams{
+		TTL:      DefaultTTL,
+		Src:      self.privatekey,
+		Dst:      &topubkey,
+		Topic:    topic,
+		WorkTime: defaultWhisperWorkTime,
+		PoW:      defaultWhisperPoW,
+		Payload:  msg,
+	}
+
+	// set up outgoing message container, which does encryption and envelope wrapping
+	woutmsg, err := whisper.NewSentMessage(wparams)
+	if err != nil {
+		return fmt.Errorf("%v: %s", errorWhisper, err)
+	}
+
+	// performs encryption and PoW
+	// after this the message is ready for sending
+	env, err := woutmsg.Wrap(wparams)
+	if err != nil {
+		return fmt.Errorf("%v: %s", errorWhisper, err)
+	}
+	log.Trace("pssmsg whisper done", "env", env)
+
 	pssmsg := &PssMsg{
 		To:      to,
-		Payload: pssenv,
+		Payload: env,
 	}
 	return self.Forward(pssmsg)
 }
@@ -298,7 +362,8 @@ func (self *Pss) SendRaw(to []byte, topic Topic, msg []byte) error {
 func (self *Pss) Forward(msg *PssMsg) error {
 
 	if self.isSelfRecipient(msg) {
-		return errorForwardToSelf
+		//return errorForwardToSelf
+		return self.Process(msg)
 	}
 
 	// cache it
@@ -360,7 +425,7 @@ func (self *Pss) Forward(msg *PssMsg) error {
 // Links a remote peer and Topic to a dedicated p2p.MsgReadWriter in the pss peerpool, and runs the specificed protocol using these resources.
 //
 // The effect is that now we have a "virtual" protocol running on an artificial p2p.Peer, which can be looked up and piped to through Pss using swarm overlay address and topic
-func (self *Pss) AddPeer(p *p2p.Peer, addr pot.Address, run func(*p2p.Peer, p2p.MsgReadWriter) error, topic Topic, rw p2p.MsgReadWriter) error {
+func (self *Pss) AddPeer(p *p2p.Peer, addr pot.Address, run func(*p2p.Peer, p2p.MsgReadWriter) error, topic whisper.TopicType, rw p2p.MsgReadWriter) error {
 	self.lock.Lock()
 	defer self.lock.Unlock()
 	self.addPeerTopic(addr, topic, rw)
@@ -372,15 +437,15 @@ func (self *Pss) AddPeer(p *p2p.Peer, addr pot.Address, run func(*p2p.Peer, p2p.
 	return nil
 }
 
-func (self *Pss) addPeerTopic(id pot.Address, topic Topic, rw p2p.MsgReadWriter) error {
+func (self *Pss) addPeerTopic(id pot.Address, topic whisper.TopicType, rw p2p.MsgReadWriter) error {
 	if self.peerPool[id] == nil {
-		self.peerPool[id] = make(map[Topic]p2p.MsgReadWriter, PssPeerTopicDefaultCapacity)
+		self.peerPool[id] = make(map[whisper.TopicType]p2p.MsgReadWriter, PssPeerTopicDefaultCapacity)
 	}
 	self.peerPool[id][topic] = rw
 	return nil
 }
 
-func (self *Pss) removePeerTopic(rw p2p.MsgReadWriter, topic Topic) {
+func (self *Pss) removePeerTopic(rw p2p.MsgReadWriter, topic whisper.TopicType) {
 	prw, ok := rw.(*PssReadWriter)
 	if !ok {
 		return
@@ -395,7 +460,7 @@ func (self *Pss) isSelfRecipient(msg *PssMsg) bool {
 	return bytes.Equal(msg.To, self.Overlay.BaseAddr())
 }
 
-func (self *Pss) isActive(id pot.Address, topic Topic) bool {
+func (self *Pss) isActive(id pot.Address, topic whisper.TopicType) bool {
 	if self.peerPool[id] == nil {
 		return false
 	}
@@ -413,7 +478,7 @@ type PssReadWriter struct {
 	LastActive time.Time
 	rw         chan p2p.Msg
 	spec       *protocols.Spec
-	topic      *Topic
+	topic      *whisper.TopicType
 }
 
 // Implements p2p.MsgReader
@@ -436,7 +501,7 @@ func (prw PssReadWriter) WriteMsg(msg p2p.Msg) error {
 	if err != nil {
 		return err
 	}
-	return prw.SendRaw(prw.To.Bytes(), *prw.topic, pmsg)
+	return prw.SendAsym(prw.To.Bytes(), *prw.topic, pmsg)
 }
 
 // Injects a p2p.Msg into the MsgReadWriter, so that it appears on the associated p2p.MsgReader
@@ -452,14 +517,14 @@ func (prw PssReadWriter) injectMsg(msg p2p.Msg) error {
 type PssProtocol struct {
 	*Pss
 	proto *p2p.Protocol
-	topic *Topic
+	topic *whisper.TopicType
 	spec  *protocols.Spec
 }
 
 // For devp2p protocol integration only.
 //
 // Maps a Topic to a devp2p protocol.
-func RegisterPssProtocol(ps *Pss, topic *Topic, spec *protocols.Spec, targetprotocol *p2p.Protocol) *PssProtocol {
+func RegisterPssProtocol(ps *Pss, topic *whisper.TopicType, spec *protocols.Spec, targetprotocol *p2p.Protocol) *PssProtocol {
 	pp := &PssProtocol{
 		Pss:   ps,
 		proto: targetprotocol,
@@ -497,3 +562,7 @@ func (self *PssProtocol) Handle(msg []byte, p *p2p.Peer, senderAddr []byte) erro
 
 	return nil
 }
+
+func getPadding() []byte {
+	return []byte{0x64, 0x6f, 0x6f, 0x62, 0x61, 0x72}
+}

swarm/pss/pssapi.go

@@ -1,14 +1,15 @@
 package pss
 
 import (
-	"bytes"
 	"context"
+	"crypto/ecdsa"
 	"fmt"
 
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/p2p"
+	"github.com/ethereum/go-ethereum/pot"
 	"github.com/ethereum/go-ethereum/rpc"
-	"github.com/ethereum/go-ethereum/swarm/network"
+	whisper "github.com/ethereum/go-ethereum/whisper/whisperv5"
 )
 
 // Pss API services
@@ -25,25 +26,25 @@ func NewAPI(ps *Pss) *API {
 // A new handler is registered in pss for the supplied topic
 //
 // All incoming messages to the node matching this topic will be encapsulated in the APIMsg struct and sent to the subscriber
-func (pssapi *API) Receive(ctx context.Context, topic Topic) (*rpc.Subscription, error) {
+func (pssapi *API) Receive(ctx context.Context, topic whisper.TopicType) (*rpc.Subscription, error) {
 	notifier, supported := rpc.NotifierFromContext(ctx)
 	if !supported {
 		return nil, fmt.Errorf("Subscribe not supported")
 	}
 
 	psssub := notifier.CreateSubscription()
+
 	handler := func(msg []byte, p *p2p.Peer, from []byte) error {
 		apimsg := &APIMsg{
 			Msg:  msg,
 			Addr: from,
 		}
 		if err := notifier.Notify(psssub.ID, apimsg); err != nil {
-			log.Warn(fmt.Sprintf("notification on pss sub topic %v rpc (sub %v) msg %v failed!", topic, psssub.ID, msg))
+			log.Warn(fmt.Sprintf("notification on pss sub topic rpc (sub %v) msg %v failed!", psssub.ID, msg))
 		}
 		return nil
 	}
 	deregf := pssapi.Register(&topic, handler)
-
 	go func() {
 		defer deregf()
 		select {
@@ -62,18 +63,8 @@ func (pssapi *API) Receive(ctx context.Context, topic Topic) (*rpc.Subscription,
 // Wrapper method for the pss.SendRaw function.
 //
 // The method will pass on the error received from pss.
-//
+func (pssapi *API) Send(topic whisper.TopicType, msg APIMsg) error {
-// Note that normally pss will report an error if an attempt is made to send a pss to oneself. However, if the debug flag has been set, and the address specified in APIMsg is the node's own, this method implements a short-circuit which injects the message as an incoming message (using Pss.Process). This can be useful for testing purposes, when only operating with one node.
+	return pssapi.SendAsym(msg.Addr, topic, msg.Msg)
-func (pssapi *API) Send(topic Topic, msg APIMsg) error {
-	if pssapi.debug && bytes.Equal(msg.Addr, pssapi.Pss.BaseAddr()) {
-		log.Warn("Pss debug enabled; send to self shortcircuit", "apimsg", msg, "topic", topic)
-		env := NewEnvelope(msg.Addr, topic, msg.Msg)
-		return pssapi.Process(&PssMsg{
-			To:      pssapi.Pss.BaseAddr(),
-			Payload: env,
-		})
-	}
-	return pssapi.SendRaw(msg.Addr, topic, msg.Msg)
 }
 
 // BaseAddr returns the pss node's swarm overlay address
@@ -83,6 +74,13 @@ func (pssapi *API) BaseAddr() ([]byte, error) {
 	return pssapi.Pss.BaseAddr(), nil
 }
 
+func (pssapi *API) AddAddressKeyPair(addr []byte, pubkey ecdsa.PublicKey) error {
+	var potaddr pot.Address
+	copy(potaddr[:], addr)
+	pssapi.Pss.AddAddressKeyPair(potaddr, pubkey)
+	return nil
+}
+
 // PssAPITest are temporary API calls for development use only
 //
 // These symbols should NOT be included in production environment
@@ -94,20 +92,3 @@ type APITest struct {
 func NewAPITest(ps *Pss) *APITest {
 	return &APITest{Pss: ps}
 }
-
-// Get the current nearest swarm node to the specified address
-//
-// (Can be used for diagnosing kademlia state)
-func (pssapitest *APITest) GetForwarder(addr []byte) (fwd struct {
-	Addr  []byte
-	Count int
-}) {
-	pssapitest.Overlay.EachConn(addr, 255, func(op network.OverlayConn, po int, isproxbin bool) bool {
-		if bytes.Equal(fwd.Addr, []byte{}) {
-			fwd.Addr = op.Address()
-		}
-		fwd.Count++
-		return true
-	})
-	return
-}

swarm/pss/types.go

@@ -2,6 +2,7 @@ package pss
 
 import (
 	"bytes"
+	"crypto/ecdsa"
 	"encoding/binary"
 	"fmt"
 	"time"
@@ -10,32 +11,35 @@ import (
 	"github.com/ethereum/go-ethereum/crypto/sha3"
 	"github.com/ethereum/go-ethereum/p2p"
 	"github.com/ethereum/go-ethereum/rlp"
+	whisper "github.com/ethereum/go-ethereum/whisper/whisperv5"
 )
 
 const (
-	TopicLength           = 32
+	TopicLength            = 32
-	DefaultTTL            = 6000
+	DefaultTTL             = 6000
-	defaultDigestCacheTTL = time.Second
+	defaultDigestCacheTTL  = time.Second
+	defaultWhisperWorkTime = 1
+	defaultWhisperPoW      = 0.01
 )
 
 // Pss configuration parameters
 type PssParams struct {
-	Cachettl time.Duration
+	Cachettl   time.Duration
-	Debug    bool
+	privatekey *ecdsa.PrivateKey
 }
 
 // Sane defaults for Pss
-func NewPssParams(debug bool) *PssParams {
+func NewPssParams(privatekey *ecdsa.PrivateKey) *PssParams {
 	return &PssParams{
-		Cachettl: defaultDigestCacheTTL,
+		Cachettl:   defaultDigestCacheTTL,
-		Debug:    debug,
+		privatekey: privatekey,
 	}
 }
 
 // Encapsulates messages transported over pss.
 type PssMsg struct {
 	To      []byte
-	Payload *Envelope
+	Payload *whisper.Envelope
 }
 
 // serializes the message for use in cache
@@ -49,24 +53,6 @@ func (self *PssMsg) String() string {
 	return fmt.Sprintf("PssMsg: Recipient: %x", common.ByteLabel(self.To))
 }
 
-// Pre-Whisper placeholder, payload of PssMsg, sender address, Topic
-type Envelope struct {
-	Topic   Topic
-	TTL     uint16
-	Payload []byte
-	From    []byte
-}
-
-// Creates A Pss envelope from sender address, topic and raw payload
-func NewEnvelope(addr []byte, topic Topic, payload []byte) *Envelope {
-	return &Envelope{
-		From:    addr,
-		Topic:   topic,
-		TTL:     DefaultTTL,
-		Payload: payload,
-	}
-}
-
 // Convenience wrapper for devp2p protocol messages for transport over pss
 type ProtocolMsg struct {
 	Code       uint64
@@ -109,26 +95,22 @@ func (self *APIMsg) String() string {
 // Implementations of this type are passed to Pss.Register together with a topic,
 type Handler func(msg []byte, p *p2p.Peer, from []byte) error
 
-// Topic defines the context of a message being transported over pss
-// It is used by pss to determine what action is to be taken on an incoming message
-// Typically, one can map protocol handlers for the message payloads by mapping topic to them; see Pss.Register
-type Topic [TopicLength]byte
-
 // String representation of Topic
-func (self *Topic) String() string {
+//func (self *Topic) String() string {
-	return fmt.Sprintf("%x", self)
+//	return fmt.Sprintf("%x", self)
-}
+//}
 
 // Constructs a new PssTopic from a given name and version.
 //
 // Analogous to the name and version members of p2p.Protocol.
-func NewTopic(s string, v int) (topic Topic) {
+func NewTopic(s string, v int) (topic whisper.TopicType) {
 	h := sha3.NewKeccak256()
 	h.Write([]byte(s))
 	buf := make([]byte, TopicLength/8)
 	binary.PutUvarint(buf, uint64(v))
 	h.Write(buf)
-	copy(topic[:], h.Sum(buf)[:])
+	//copy(topic[:], h.Sum(buf)[:])
+	topic = whisper.BytesToTopic(h.Sum(buf)[:4])
 	return topic
 }
 

swarm/swarm.go

@@ -142,7 +142,7 @@ func NewSwarm(ctx *node.ServiceContext, backend chequebook.Backend, config *api.
 
 	// Pss = postal service over swarm (devp2p over bzz)
 	if pssEnabled {
-		pssparams := pss.NewPssParams(false)
+		pssparams := pss.NewPssParams(self.privateKey)
 		self.pss = pss.NewPss(to, self.dpa, pssparams)
 	}
 

whisper/whisperv5/topic.go

@@ -47,7 +47,7 @@ func (topic *TopicType) String() string {
 }
 
 // UnmarshalJSON parses a hex representation to a topic.
-func (t *TopicType) UnmarshalJSON(input []byte) error {
+func (t *TopicType) UnmarshalJSON_(input []byte) error {
 	length := len(input)
 	if length >= 2 && input[0] == '"' && input[length-1] == '"' {
 		input = input[1 : length-1]
@@ -60,6 +60,7 @@ func (t *TopicType) UnmarshalJSON(input []byte) error {
 	if len(input) != TopicLength*2 {
 		return fmt.Errorf("unmarshalJSON failed: topic must be exactly %d bytes", TopicLength)
 	}
+
 	b := common.FromHex(string(input))
 	if b == nil {
 		return fmt.Errorf("unmarshalJSON failed: wrong topic format")