7993

Committed 24 Sep 2020 09:45PM UTC coverage: 40.634%. First build

Build # 7993

Build Type

Pull #2044

travis-ci

Committed by

web-flow

Commit Message

<a href="https://github.com/OpenBazaar/openbazaar-go/commit/<a class=hub.com/OpenBazaar/openbazaar-go/commit/<a class="double-link" href="https://git"><a class=hub.com/OpenBazaar/openbazaar-go/commit/<a class="double-link" href="https://git"><a class=hub.com/OpenBazaar/openbazaar-go/commit/88b39e2efd8a3b0d37c2bd5071d0c633513a862d">88b39e2ef<a href="https://github.com/OpenBazaar/openbazaar-go/commit/88b39e2efd8a3b0d37c2bd5071d0c633513a862d">&quot;&gt;&amp;lt;a href=&amp;quot;https://github.com/OpenBazaar/openbazaar-go/commit/&lt;/a&gt;&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/OpenBazaar/openbazaar-go/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/OpenBazaar/openbazaar-go/commit/88b39e2efd8a3b0d37c2bd5071d0c633513a862d&quot;&gt;88b39e2ef&lt;/a&gt;&lt;a href=&quot;https://github.com/OpenBazaar/openbazaar-go/commit/88b39e2efd8a3b0d37c2bd5071d0c633513a862d&quot;&gt;&amp;lt;a href=&amp;quot;https://github.com/OpenBazaar/openbazaar-go/commit/88b39e2efd8a3b0d37c2bd5071d0c633&lt;/a&gt;513a862d&quot;>&quot;&gt;Merge &lt;/a&gt;&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/OpenBazaar/openbazaar-go/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/OpenBazaar/openbazaar-go/commit/<a class="double-link" href="https://github.com/OpenBazaar/openbazaar-go/commit/431911aacfecf41f043c223e085b5c4d95a6cde6">431911aac</a><a href="https://github.com/OpenBazaar/openbazaar-go/commit/88b39e2efd8a3b0d37c2bd5071d0c633513a862d">&amp;quot;&amp;gt;431911aac&amp;lt;/a&amp;gt;&amp;quot;&amp;gt;431911aac&amp;lt;/a&amp;gt;&amp;lt;a href=&amp;quot;https://github.com/OpenBazaar/openbazaar-go/commit/88b39e2efd8a3b0d37c2bd5071d0c633513a8&lt;/a&gt;62d&quot;&gt; into &lt;/a&gt;&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/OpenBazaar/openbazaar-go/commit/<a class="double-link" href="https://github.com/OpenBazaar/openbazaar-go/commit/b4be594fc">b4be594fc">b4be594fc</a>

Pull Request Pull Request #2044: Ethereum into Mobile

Run Details

20 of 20 new or added lines in 3 files covered. (100.0%)

10799 of 26576 relevant lines covered (40.63%)

0.43 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

0.0

/net/retriever/retriever.go

package net

import (
        "context"
        "errors"

        routing "gx/ipfs/QmSY3nkMNLzh9GdbFKK5tT7YMfLpf52iUZ8ZRkr29MJaa5/go-libp2p-kad-dht"
        libp2p "gx/ipfs/QmTW4SdgBWq9GjsBsHeUx8WuGxzhgzAf88UMH2w62PC8yK/go-libp2p-crypto"
        ma "gx/ipfs/QmTZBfrPJmjWsCvHEtX5FE6KimVJhsJg5sBbqEFYf4UZtL/go-multiaddr"
        "gx/ipfs/QmTbxNB1NwDesLmKTscr4udL2tVP7MaxvXnD1D9yX7g3PN/go-cid"
        "gx/ipfs/QmYVXrKrKHDC9FobgmcmshCDyWwdrfwfanNQN4oxJ9Fk3h/go-libp2p-peer"
        ps "gx/ipfs/QmaCTz9RkrU13bm9kMB54f7atgqM4qkjDZpRwRoJiWXEqs/go-libp2p-peerstore"
        "gx/ipfs/QmerPMzPk1mJVowm8KgmoknWa4yCYvvugMPsgWmDNUvDLW/go-multihash"

        "io/ioutil"
        "net/http"
        "sync"
        "time"

        "github.com/OpenBazaar/openbazaar-go/ipfs"
        "github.com/OpenBazaar/openbazaar-go/net"
        "github.com/OpenBazaar/openbazaar-go/pb"
        "github.com/OpenBazaar/openbazaar-go/repo"
        "github.com/golang/protobuf/proto"
        "github.com/ipfs/go-ipfs/core"
        "github.com/op/go-logging"
        "golang.org/x/net/proxy"
)

const DefaultPointerPrefixLength = 14

var (
        // Initialize a clear pointerList for the DHT on start
        pointerList = []string{}
        log         = logging.MustGetLogger("retriever")
)

type MRConfig struct {
        Db        repo.Datastore
        IPFSNode  *core.IpfsNode
        DHT       *routing.IpfsDHT
        BanManger *net.BanManager
        Service   net.NetworkService
        PrefixLen int
        PushNodes []peer.ID
        Dialer    proxy.Dialer
        SendAck   func(peerId string, pointerID peer.ID) error
        SendError func(peerId string, k *libp2p.PubKey, errorMessage pb.Message) error
}

type MessageRetriever struct {
        db         repo.Datastore
        node       *core.IpfsNode
        routing    *routing.IpfsDHT
        bm         *net.BanManager
        service    net.NetworkService
        prefixLen  int
        sendAck    func(peerId string, pointerID peer.ID) error
        sendError  func(peerId string, k *libp2p.PubKey, errorMessage pb.Message) error
        httpClient *http.Client
        dataPeers  []peer.ID
        queueLock  *sync.Mutex
        DoneChan   chan struct{}
        inFlight   chan struct{}
        *sync.WaitGroup
}

type offlineMessage struct {
        addr string
        env  pb.Envelope
}

func stringInSlice(str string, list []string) bool {
        for _, v := range list {
                if v == str {
                        return true
                }
        }
        return false
}

// Reset on startup
func (m *MessageRetriever) ResetPointerList() {
        pointerList = []string{}
}

func NewMessageRetriever(cfg MRConfig) *MessageRetriever {
        var client *http.Client
        if cfg.Dialer != nil {
                tbTransport := &http.Transport{Dial: cfg.Dialer.Dial}
                client = &http.Client{Transport: tbTransport, Timeout: time.Second * 30}
        } else {
                client = &http.Client{Timeout: time.Second * 30}
        }
        mr := MessageRetriever{
                db:         cfg.Db,
                node:       cfg.IPFSNode,
                routing:    cfg.DHT,
                bm:         cfg.BanManger,
                service:    cfg.Service,
                prefixLen:  cfg.PrefixLen,
                sendAck:    cfg.SendAck,
                sendError:  cfg.SendError,
                httpClient: client,
                dataPeers:  cfg.PushNodes,
                queueLock:  new(sync.Mutex),
                DoneChan:   make(chan struct{}),
                inFlight:   make(chan struct{}, 5),
                WaitGroup:  new(sync.WaitGroup),
        }

        mr.Add(2)
        return &mr
}

func (m *MessageRetriever) Run() {
        dht := time.NewTicker(time.Hour)
        peers := time.NewTicker(time.Minute)
        defer dht.Stop()
        defer peers.Stop()
        go m.fetchPointersFromPushNodes()
        go m.fetchPointersFromDHT()
        for {
                select {
                case <-dht.C:
                        m.Add(1)
                        go m.fetchPointersFromDHT()
                case <-peers.C:
                        m.Add(1)
                        go m.fetchPointersFromPushNodes()
                }
        }
}

// RunOnce - used to fetch messages only once
func (m *MessageRetriever) RunOnce() {
        m.Add(1)
        go m.fetchPointersFromDHT()
        m.Add(1)
        go m.fetchPointersFromPushNodes()
}

func (m *MessageRetriever) fetchPointersFromDHT() {
        ctx, cancel := context.WithCancel(context.Background())
        defer cancel()
        mh, _ := multihash.FromB58String(m.node.Identity.Pretty())
        peerOut := make(chan ps.PeerInfo)
        go func(c chan ps.PeerInfo) {
                iout := ipfs.FindPointersAsync(m.routing, ctx, mh, m.prefixLen)
                for p := range iout {
                        c <- p
                }
                close(c)

        }(peerOut)

        m.downloadMessages(peerOut)
}

func (m *MessageRetriever) fetchPointersFromPushNodes() {
        peerOut := make(chan ps.PeerInfo)
        go func(c chan ps.PeerInfo) {
                out := m.getPointersDataPeers()
                for p := range out {
                        c <- p
                }
                close(c)

        }(peerOut)
        m.downloadMessages(peerOut)
}

func (m *MessageRetriever) downloadMessages(peerOut chan ps.PeerInfo) {
        wg := new(sync.WaitGroup)
        downloaded := 0

        inFlight := make(map[string]bool)
        // Iterate over the pointers, adding 1 to the waitgroup for each pointer found
        for p := range peerOut {
                if len(p.Addrs) > 0 && !m.db.OfflineMessages().Has(p.Addrs[0].String()) && !stringInSlice(p.Addrs[0].String(), pointerList) && !inFlight[p.Addrs[0].String()] {
                        pointerList = append(pointerList, p.Addrs[0].String())
                        log.Debugf("Looking for pointer [%v] at %v\n", p.ID.Pretty(), p.Addrs)
                        inFlight[p.Addrs[0].String()] = true
                        log.Debugf("Found pointer with location %s", p.Addrs[0].String())
                        // IPFS
                        if len(p.Addrs[0].Protocols()) == 1 && p.Addrs[0].Protocols()[0].Code == ma.P_IPFS {
                                wg.Add(1)
                                downloaded++
                                go m.fetchIPFS(p.ID, m.node, p.Addrs[0], wg)
                        }

                        // HTTPS
                        if len(p.Addrs[0].Protocols()) == 2 && p.Addrs[0].Protocols()[0].Code == ma.P_IPFS && p.Addrs[0].Protocols()[1].Code == ma.P_HTTPS {
                                enc, err := p.Addrs[0].ValueForProtocol(ma.P_IPFS)
                                if err != nil {
                                        continue
                                }
                                mh, err := multihash.FromB58String(enc)
                                if err != nil {
                                        continue
                                }
                                d, err := multihash.Decode(mh)
                                if err != nil {
                                        continue
                                }
                                wg.Add(1)
                                downloaded++
                                go m.fetchHTTPS(p.ID, string(d.Digest), p.Addrs[0], wg)
                        }
                }
        }

        // Wait for each goroutine to finish then process any remaining messages that needed to be processed last
        wg.Wait()

        m.processQueuedMessages()

        m.Done()
}

// Connect directly to our data peers and ask them if they have the pointer we're interested in
func (m *MessageRetriever) getPointersDataPeers() <-chan ps.PeerInfo {
        peerOut := make(chan ps.PeerInfo, 100000)
        go m.getPointersFromDataPeersRoutine(peerOut)
        return peerOut
}

func (m *MessageRetriever) getPointersFromDataPeersRoutine(peerOut chan ps.PeerInfo) {
        defer close(peerOut)
        mh, _ := multihash.FromB58String(m.node.Identity.Pretty())
        keyhash := ipfs.CreatePointerKey(mh, DefaultPointerPrefixLength)
        k, _ := cid.Decode(keyhash.B58String())
        var wg sync.WaitGroup
        for _, p := range m.dataPeers {
                wg.Add(1)
                go func(pid peer.ID) {
                        defer wg.Done()
                        ctx, cancel := context.WithTimeout(context.Background(), time.Second*35)
                        defer cancel()
                        time.Sleep(time.Second * 15)
                        provs, err := ipfs.GetPointersFromPeer(m.routing, ctx, pid, &k)
                        if err != nil {
                                log.Errorf("Could not get pointers from push node because: %v", err)
                                return
                        }
                        log.Debugf("Successfully queried %s for pointers", pid.Pretty())
                        for _, pi := range provs {
                                peerOut <- *pi
                        }
                }(p)
        }
        wg.Wait()
}

// fetchIPFS will attempt to download an encrypted message using IPFS. If the message downloads successfully, we save the
// address to the database to prevent us from wasting bandwidth downloading it again.
func (m *MessageRetriever) fetchIPFS(pid peer.ID, n *core.IpfsNode, addr ma.Multiaddr, wg *sync.WaitGroup) {
        m.inFlight <- struct{}{}
        defer func() {
                wg.Done()
                <-m.inFlight
        }()

        c := make(chan struct{})
        var ciphertext []byte
        var err error

        go func() {
                ciphertext, err = ipfs.Cat(n, addr.String(), time.Second*10)
                c <- struct{}{}
        }()

        select {
        case <-c:
                if err != nil {
                        log.Errorf("Error retrieving offline message from: %s, Error: %s", addr.String(), err.Error())
                        return
                }
                log.Debugf("Successfully downloaded offline message %s from: %s", addr.String(), pid.Pretty())

                err = m.db.OfflineMessages().Put(addr.String())
                if err != nil {
                        log.Error(err)
                }
                m.attemptDecrypt(ciphertext, pid, addr)
        case <-m.DoneChan:
                return
        }
}

// fetchHTTPS will attempt to download an encrypted message from an HTTPS endpoint. If the message downloads successfully, we save the
// address to the database to prevent us from wasting bandwidth downloading it again.
func (m *MessageRetriever) fetchHTTPS(pid peer.ID, url string, addr ma.Multiaddr, wg *sync.WaitGroup) {
        m.inFlight <- struct{}{}
        defer func() {
                wg.Done()
                <-m.inFlight
        }()

        c := make(chan struct{})
        var ciphertext []byte
        var err error

        go func() {
                var resp *http.Response
                resp, err = m.httpClient.Get(url)
                if err != nil {
                        log.Errorf("Error retrieving offline message from %s, %s", addr.String(), err.Error())
                        c <- struct{}{}
                        return
                }
                ciphertext, err = ioutil.ReadAll(resp.Body)
        }()

        select {
        case <-c:
                if err != nil {
                        log.Errorf("Error retrieving offline message from %s, %s", addr.String(), err.Error())
                        return
                }
                log.Debugf("Successfully downloaded offline message from %s", addr.String())
                err = m.db.OfflineMessages().Put(addr.String())
                if err != nil {
                        log.Error(err)
                }
                m.attemptDecrypt(ciphertext, pid, addr)
        case <-m.DoneChan:
                return
        }
}

// attemptDecrypt will try to decrypt the message using our identity private key. If it decrypts it will be passed to
// a handler for processing. Not all messages will decrypt. Given the natural of the prefix addressing, we may download
// some messages intended for others. If we can't decrypt it, we can just discard it.
func (m *MessageRetriever) attemptDecrypt(ciphertext []byte, pid peer.ID, addr ma.Multiaddr) {
        // Decrypt and unmarshal plaintext
        plaintext, err := net.Decrypt(m.node.PrivateKey, ciphertext)
        if err != nil {
                log.Warningf("Unable to decrypt cipher text to plain text, CID: %s: Error:%s\n", addr.String(), err.Error())
                return
        }

        // Unmarshal plaintext
        env := pb.Envelope{}
        err = proto.Unmarshal(plaintext, &env)
        if err != nil {
                log.Warningf("Unable to unmarshal plaintext to encrypted Envelope, CID: %s: Error:%s\n", addr.String(), err.Error())
                return
        }

        // Validate the signature
        ser, err := proto.Marshal(env.Message)
        if err != nil {
                log.Warningf("Unable to serialize the encrypted message, CID: %s: Error:%s\n", addr.String(), err.Error())
                return
        }
        pubkey, err := libp2p.UnmarshalPublicKey(env.Pubkey)
        if err != nil {
                log.Warningf("Unable to unmarshal the public key from, CID: %s: Error:%s\n", addr.String(), err.Error())
                return
        }

        valid, err := pubkey.Verify(ser, env.Signature)
        if err != nil || !valid {
                log.Warningf("Unable to verify message signature, CID: %s: Error:%s\n", addr.String(), err.Error())
                return
        }

        id, err := peer.IDFromPublicKey(pubkey)
        if err != nil {
                log.Warningf("Unable to get a peer ID from the pubkey, CID: %s: Error:%s\n", addr.String(), err.Error())
                return
        }

        log.Debugf("Received offline message %s from: %s\n", addr.String(), id.Pretty())

        if m.bm.IsBanned(id) {
                log.Warningf("Received and dropped offline message from banned user: %s\n", id.Pretty())
                return
        }

        if err := m.node.Peerstore.AddPubKey(id, pubkey); err != nil {
                log.Errorf("adding pubkey to peerstore: %s", err.Error())
        }
        store := m.node.Repo.Datastore()
        if err := ipfs.PutCachedPubkey(store, id.Pretty(), env.Pubkey); err != nil {
                log.Errorf("caching pubkey: %s", err.Error())
        }

        // Respond with an ACK
        if env.Message.MessageType != pb.Message_OFFLINE_ACK {
                err = m.sendAck(id.Pretty(), pid)
                if err != nil {
                        log.Error(err)
                }
        }

        // handle
        err = m.handleMessage(env, addr.String(), nil)
        if err != nil {
                log.Error(err)
        }
}

// handleMessage loads the handler for this message type and attempts to process the message. Some message types (such
// as those partaining to an order) need to be processed in order. In these cases the handler returns a net.OutOfOrderMessage error
// and we must save the message to the database to await further processing.
func (m *MessageRetriever) handleMessage(env pb.Envelope, addr string, id *peer.ID) error {
        if id == nil {
                // Get the peer ID from the public key
                pubkey, err := libp2p.UnmarshalPublicKey(env.Pubkey)
                if err != nil {
                        log.Errorf("Error processing message %s. Type %s: %s", addr, env.Message.MessageType, err.Error())
                        return err
                }
                i, err := peer.IDFromPublicKey(pubkey)
                if err != nil {
                        log.Errorf("Error processing message %s. Type %s: %s", addr, env.Message.MessageType, err.Error())
                        return err
                }
                id = &i
        }

        // Get handler for this message type
        handler := m.service.HandlerForMsgType(env.Message.MessageType)
        if handler == nil {
                log.Errorf("Nil handler for message type %s", env.Message.MessageType)
                return errors.New("nil handler for message")
        }

        // Dispatch handler
        resp, err := handler(*id, env.Message, true)
        if err != nil {
                if err == net.OutOfOrderMessage {
                        ser, err := proto.Marshal(&env)
                        if err == nil {
                                err := m.db.OfflineMessages().SetMessage(addr, ser)
                                if err != nil {
                                        log.Errorf("Error saving offline message %s to database: %s", addr, err.Error())
                                }
                        } else {
                                log.Errorf("Error serializing offline message %s for storage", addr)
                        }
                } else if env.Message.MessageType == pb.Message_ORDER && resp != nil {
                        log.Errorf("Error processing ORDER message: %s, sending ERROR response", err.Error())
                        err = m.sendError(id.Pretty(), nil, *resp)
                        if err != nil {
                                log.Error(err)
                        }
                        return err
                } else {
                        log.Errorf("Error processing message %s. Type %s: %s", addr, env.Message.MessageType, err.Error())
                        return err
                }
        }
        return nil
}

var MessageProcessingOrder = []pb.Message_MessageType{
        pb.Message_ORDER,
        pb.Message_ORDER_CANCEL,
        pb.Message_ORDER_REJECT,
        pb.Message_ORDER_CONFIRMATION,
        pb.Message_ORDER_PAYMENT,
        pb.Message_ORDER_FULFILLMENT,
        pb.Message_ORDER_COMPLETION,
        pb.Message_DISPUTE_OPEN,
        pb.Message_DISPUTE_UPDATE,
        pb.Message_VENDOR_FINALIZED_PAYMENT,
        pb.Message_DISPUTE_CLOSE,
        pb.Message_REFUND,
        pb.Message_CHAT,
        pb.Message_FOLLOW,
        pb.Message_UNFOLLOW,
        pb.Message_MODERATOR_ADD,
        pb.Message_MODERATOR_REMOVE,
        pb.Message_OFFLINE_ACK,
        pb.Message_OFFLINE_RELAY,
}

// processQueuedMessages loads all the saved messaged from the database for processing. For each message it sorts them into a
// queue based on message type and then processes the queue in order. Any messages that successfully process can then be deleted
// from the database.
func (m *MessageRetriever) processQueuedMessages() {
        messageQueue := make(map[pb.Message_MessageType][]offlineMessage)
        for _, messageType := range MessageProcessingOrder {
                messageQueue[messageType] = []offlineMessage{}
        }

        // Load stored messages from database
        messages, err := m.db.OfflineMessages().GetMessages()
        if err != nil {
                return
        }
        // Sort them into the queue by message type
        for url, ser := range messages {
                env := new(pb.Envelope)
                err := proto.Unmarshal(ser, env)
                if err == nil {
                        messageQueue[env.Message.MessageType] = append(messageQueue[env.Message.MessageType], offlineMessage{url, *env})
                } else {
                        log.Error("Error unmarshalling serialized offline message from database")
                }
        }
        var toDelete []string
        // Process the queue in order
        for _, messageType := range MessageProcessingOrder {
                queue, ok := messageQueue[messageType]
                if !ok {
                        continue
                }
                for _, om := range queue {
                        err := m.handleMessage(om.env, om.addr, nil)
                        if err == nil {
                                toDelete = append(toDelete, om.addr)
                        }
                }
        }
        // Delete messages that we're successfully processed from the database
        for _, url := range toDelete {
                err = m.db.OfflineMessages().DeleteMessage(url)
                if err != nil {
                        log.Error(err)
                }
        }
}

1	package net
2
3	import (
4	"context"
5	"errors"
6
7	routing "gx/ipfs/QmSY3nkMNLzh9GdbFKK5tT7YMfLpf52iUZ8ZRkr29MJaa5/go-libp2p-kad-dht"
8	libp2p "gx/ipfs/QmTW4SdgBWq9GjsBsHeUx8WuGxzhgzAf88UMH2w62PC8yK/go-libp2p-crypto"
9	ma "gx/ipfs/QmTZBfrPJmjWsCvHEtX5FE6KimVJhsJg5sBbqEFYf4UZtL/go-multiaddr"
10	"gx/ipfs/QmTbxNB1NwDesLmKTscr4udL2tVP7MaxvXnD1D9yX7g3PN/go-cid"
11	"gx/ipfs/QmYVXrKrKHDC9FobgmcmshCDyWwdrfwfanNQN4oxJ9Fk3h/go-libp2p-peer"
12	ps "gx/ipfs/QmaCTz9RkrU13bm9kMB54f7atgqM4qkjDZpRwRoJiWXEqs/go-libp2p-peerstore"
13	"gx/ipfs/QmerPMzPk1mJVowm8KgmoknWa4yCYvvugMPsgWmDNUvDLW/go-multihash"
14
15	"io/ioutil"
16	"net/http"
17	"sync"
18	"time"
19
20	"github.com/OpenBazaar/openbazaar-go/ipfs"
21	"github.com/OpenBazaar/openbazaar-go/net"
22	"github.com/OpenBazaar/openbazaar-go/pb"
23	"github.com/OpenBazaar/openbazaar-go/repo"
24	"github.com/golang/protobuf/proto"
25	"github.com/ipfs/go-ipfs/core"
26	"github.com/op/go-logging"
27	"golang.org/x/net/proxy"
28	)
29
30	const DefaultPointerPrefixLength = 14
31
32	var (
33	// Initialize a clear pointerList for the DHT on start
34	pointerList = []string{}
35	log = logging.MustGetLogger("retriever")
36	)
37
38	type MRConfig struct {
39	Db repo.Datastore
40	IPFSNode *core.IpfsNode
41	DHT *routing.IpfsDHT
42	BanManger *net.BanManager
43	Service net.NetworkService
44	PrefixLen int
45	PushNodes []peer.ID
46	Dialer proxy.Dialer
47	SendAck func(peerId string, pointerID peer.ID) error
48	SendError func(peerId string, k *libp2p.PubKey, errorMessage pb.Message) error
49	}
50
51	type MessageRetriever struct {
52	db repo.Datastore
53	node *core.IpfsNode
54	routing *routing.IpfsDHT
55	bm *net.BanManager
56	service net.NetworkService
57	prefixLen int
58	sendAck func(peerId string, pointerID peer.ID) error
59	sendError func(peerId string, k *libp2p.PubKey, errorMessage pb.Message) error
60	httpClient *http.Client
61	dataPeers []peer.ID
62	queueLock *sync.Mutex
63	DoneChan chan struct{}
64	inFlight chan struct{}
65	*sync.WaitGroup
66	}
67
68	type offlineMessage struct {
69	addr string
70	env pb.Envelope
71	}
72
73	func stringInSlice(str string, list []string) bool {	×
74	for _, v := range list {	×
75	if v == str {	×
76	return true	×
77	}	×
78	}
79	return false	×
80	}
81
82	// Reset on startup
83	func (m *MessageRetriever) ResetPointerList() {	×
84	pointerList = []string{}	×
85	}	×
86
87	func NewMessageRetriever(cfg MRConfig) *MessageRetriever {	×
88	var client *http.Client	×
89	if cfg.Dialer != nil {	×
90	tbTransport := &http.Transport{Dial: cfg.Dialer.Dial}	×
91	client = &http.Client{Transport: tbTransport, Timeout: time.Second * 30}	×
92	} else {	×
93	client = &http.Client{Timeout: time.Second * 30}	×
94	}	×
95	mr := MessageRetriever{	×
96	db: cfg.Db,	×
97	node: cfg.IPFSNode,	×
98	routing: cfg.DHT,	×
99	bm: cfg.BanManger,	×
100	service: cfg.Service,	×
101	prefixLen: cfg.PrefixLen,	×
102	sendAck: cfg.SendAck,	×
103	sendError: cfg.SendError,	×
104	httpClient: client,	×
105	dataPeers: cfg.PushNodes,	×
106	queueLock: new(sync.Mutex),	×
107	DoneChan: make(chan struct{}),	×
108	inFlight: make(chan struct{}, 5),	×
109	WaitGroup: new(sync.WaitGroup),	×
110	}	×
111		×
112	mr.Add(2)	×
113	return &mr	×
114	}
115
116	func (m *MessageRetriever) Run() {	×
117	dht := time.NewTicker(time.Hour)	×
118	peers := time.NewTicker(time.Minute)	×
119	defer dht.Stop()	×
120	defer peers.Stop()	×
121	go m.fetchPointersFromPushNodes()	×
122	go m.fetchPointersFromDHT()	×
123	for {	×
124	select {	×
125	case <-dht.C:	×
126	m.Add(1)	×
127	go m.fetchPointersFromDHT()	×
128	case <-peers.C:	×
129	m.Add(1)	×
130	go m.fetchPointersFromPushNodes()	×
131	}
132	}
133	}
134
135	// RunOnce - used to fetch messages only once
136	func (m *MessageRetriever) RunOnce() {	×
137	m.Add(1)	×
138	go m.fetchPointersFromDHT()	×
139	m.Add(1)	×
140	go m.fetchPointersFromPushNodes()	×
141	}	×
142
143	func (m *MessageRetriever) fetchPointersFromDHT() {	×
144	ctx, cancel := context.WithCancel(context.Background())	×
145	defer cancel()	×
146	mh, _ := multihash.FromB58String(m.node.Identity.Pretty())	×
147	peerOut := make(chan ps.PeerInfo)	×
148	go func(c chan ps.PeerInfo) {	×
149	iout := ipfs.FindPointersAsync(m.routing, ctx, mh, m.prefixLen)	×
150	for p := range iout {	×
151	c <- p	×
152	}	×
153	close(c)	×
154
155	}(peerOut)
156
157	m.downloadMessages(peerOut)	×
158	}
159
160	func (m *MessageRetriever) fetchPointersFromPushNodes() {	×
161	peerOut := make(chan ps.PeerInfo)	×
162	go func(c chan ps.PeerInfo) {	×
163	out := m.getPointersDataPeers()	×
164	for p := range out {	×
165	c <- p	×
166	}	×
167	close(c)	×
168
169	}(peerOut)
170	m.downloadMessages(peerOut)	×
171	}
172
173	func (m *MessageRetriever) downloadMessages(peerOut chan ps.PeerInfo) {	×
174	wg := new(sync.WaitGroup)	×
175	downloaded := 0	×
176		×
177	inFlight := make(map[string]bool)	×
178	// Iterate over the pointers, adding 1 to the waitgroup for each pointer found	×
179	for p := range peerOut {	×
180	if len(p.Addrs) > 0 && !m.db.OfflineMessages().Has(p.Addrs[0].String()) && !stringInSlice(p.Addrs[0].String(), pointerList) && !inFlight[p.Addrs[0].String()] {	×
181	pointerList = append(pointerList, p.Addrs[0].String())	×
182	log.Debugf("Looking for pointer [%v] at %v\n", p.ID.Pretty(), p.Addrs)	×
183	inFlight[p.Addrs[0].String()] = true	×
184	log.Debugf("Found pointer with location %s", p.Addrs[0].String())	×
185	// IPFS	×
186	if len(p.Addrs[0].Protocols()) == 1 && p.Addrs[0].Protocols()[0].Code == ma.P_IPFS {	×
187	wg.Add(1)	×
188	downloaded++	×
189	go m.fetchIPFS(p.ID, m.node, p.Addrs[0], wg)	×
190	}	×
191
192	// HTTPS
193	if len(p.Addrs[0].Protocols()) == 2 && p.Addrs[0].Protocols()[0].Code == ma.P_IPFS && p.Addrs[0].Protocols()[1].Code == ma.P_HTTPS {	×
194	enc, err := p.Addrs[0].ValueForProtocol(ma.P_IPFS)	×
195	if err != nil {	×
196	continue	×
197	}
198	mh, err := multihash.FromB58String(enc)	×
199	if err != nil {	×
200	continue	×
201	}
202	d, err := multihash.Decode(mh)	×
203	if err != nil {	×
204	continue	×
205	}
206	wg.Add(1)	×
207	downloaded++	×
208	go m.fetchHTTPS(p.ID, string(d.Digest), p.Addrs[0], wg)	×
209	}
210	}
211	}
212
213	// Wait for each goroutine to finish then process any remaining messages that needed to be processed last
214	wg.Wait()	×
215		×
216	m.processQueuedMessages()	×
217		×
218	m.Done()	×
219	}
220
221	// Connect directly to our data peers and ask them if they have the pointer we're interested in
222	func (m *MessageRetriever) getPointersDataPeers() <-chan ps.PeerInfo {	×
223	peerOut := make(chan ps.PeerInfo, 100000)	×
224	go m.getPointersFromDataPeersRoutine(peerOut)	×
225	return peerOut	×
226	}	×
227
228	func (m *MessageRetriever) getPointersFromDataPeersRoutine(peerOut chan ps.PeerInfo) {	×
229	defer close(peerOut)	×
230	mh, _ := multihash.FromB58String(m.node.Identity.Pretty())	×
231	keyhash := ipfs.CreatePointerKey(mh, DefaultPointerPrefixLength)	×
232	k, _ := cid.Decode(keyhash.B58String())	×
233	var wg sync.WaitGroup	×
234	for _, p := range m.dataPeers {	×
235	wg.Add(1)	×
236	go func(pid peer.ID) {	×
237	defer wg.Done()	×
238	ctx, cancel := context.WithTimeout(context.Background(), time.Second*35)	×
239	defer cancel()	×
240	time.Sleep(time.Second * 15)	×
241	provs, err := ipfs.GetPointersFromPeer(m.routing, ctx, pid, &k)	×
242	if err != nil {	×
243	log.Errorf("Could not get pointers from push node because: %v", err)	×
244	return	×
245	}	×
246	log.Debugf("Successfully queried %s for pointers", pid.Pretty())	×
247	for _, pi := range provs {	×
248	peerOut <- *pi	×
249	}	×
250	}(p)
251	}
252	wg.Wait()	×
253	}
254
255	// fetchIPFS will attempt to download an encrypted message using IPFS. If the message downloads successfully, we save the
256	// address to the database to prevent us from wasting bandwidth downloading it again.
257	func (m MessageRetriever) fetchIPFS(pid peer.ID, n core.IpfsNode, addr ma.Multiaddr, wg *sync.WaitGroup) {	×
258	m.inFlight <- struct{}{}	×
259	defer func() {	×
260	wg.Done()	×
261	<-m.inFlight	×
262	}()	×
263
264	c := make(chan struct{})	×
265	var ciphertext []byte	×
266	var err error	×
267		×
268	go func() {	×
269	ciphertext, err = ipfs.Cat(n, addr.String(), time.Second*10)	×
270	c <- struct{}{}	×
271	}()	×
272
273	select {	×
274	case <-c:	×
275	if err != nil {	×
276	log.Errorf("Error retrieving offline message from: %s, Error: %s", addr.String(), err.Error())	×
277	return	×
278	}	×
279	log.Debugf("Successfully downloaded offline message %s from: %s", addr.String(), pid.Pretty())	×
280		×
281	err = m.db.OfflineMessages().Put(addr.String())	×
282	if err != nil {	×
283	log.Error(err)	×
284	}	×
285	m.attemptDecrypt(ciphertext, pid, addr)	×
286	case <-m.DoneChan:	×
287	return	×
288	}
289	}
290
291	// fetchHTTPS will attempt to download an encrypted message from an HTTPS endpoint. If the message downloads successfully, we save the
292	// address to the database to prevent us from wasting bandwidth downloading it again.
293	func (m MessageRetriever) fetchHTTPS(pid peer.ID, url string, addr ma.Multiaddr, wg sync.WaitGroup) {	×
294	m.inFlight <- struct{}{}	×
295	defer func() {	×
296	wg.Done()	×
297	<-m.inFlight	×
298	}()	×
299
300	c := make(chan struct{})	×
301	var ciphertext []byte	×
302	var err error	×
303		×
304	go func() {	×
305	var resp *http.Response	×
306	resp, err = m.httpClient.Get(url)	×
307	if err != nil {	×
308	log.Errorf("Error retrieving offline message from %s, %s", addr.String(), err.Error())	×
309	c <- struct{}{}	×
310	return	×
311	}	×
312	ciphertext, err = ioutil.ReadAll(resp.Body)	×
313	}()
314
315	select {	×
316	case <-c:	×
317	if err != nil {	×
318	log.Errorf("Error retrieving offline message from %s, %s", addr.String(), err.Error())	×
319	return	×
320	}	×
321	log.Debugf("Successfully downloaded offline message from %s", addr.String())	×
322	err = m.db.OfflineMessages().Put(addr.String())	×
323	if err != nil {	×
324	log.Error(err)	×
325	}	×
326	m.attemptDecrypt(ciphertext, pid, addr)	×
327	case <-m.DoneChan:	×
328	return	×
329	}
330	}
331
332	// attemptDecrypt will try to decrypt the message using our identity private key. If it decrypts it will be passed to
333	// a handler for processing. Not all messages will decrypt. Given the natural of the prefix addressing, we may download
334	// some messages intended for others. If we can't decrypt it, we can just discard it.
335	func (m *MessageRetriever) attemptDecrypt(ciphertext []byte, pid peer.ID, addr ma.Multiaddr) {	×
336	// Decrypt and unmarshal plaintext	×
337	plaintext, err := net.Decrypt(m.node.PrivateKey, ciphertext)	×
338	if err != nil {	×
339	log.Warningf("Unable to decrypt cipher text to plain text, CID: %s: Error:%s\n", addr.String(), err.Error())	×
340	return	×
341	}	×
342
343	// Unmarshal plaintext
344	env := pb.Envelope{}	×
345	err = proto.Unmarshal(plaintext, &env)	×
346	if err != nil {	×
347	log.Warningf("Unable to unmarshal plaintext to encrypted Envelope, CID: %s: Error:%s\n", addr.String(), err.Error())	×
348	return	×
349	}	×
350
351	// Validate the signature
352	ser, err := proto.Marshal(env.Message)	×
353	if err != nil {	×
354	log.Warningf("Unable to serialize the encrypted message, CID: %s: Error:%s\n", addr.String(), err.Error())	×
355	return	×
356	}	×
357	pubkey, err := libp2p.UnmarshalPublicKey(env.Pubkey)	×
358	if err != nil {	×
359	log.Warningf("Unable to unmarshal the public key from, CID: %s: Error:%s\n", addr.String(), err.Error())	×
360	return	×
361	}	×
362
363	valid, err := pubkey.Verify(ser, env.Signature)	×
364	if err != nil \|\| !valid {	×
365	log.Warningf("Unable to verify message signature, CID: %s: Error:%s\n", addr.String(), err.Error())	×
366	return	×
367	}	×
368
369	id, err := peer.IDFromPublicKey(pubkey)	×
370	if err != nil {	×
371	log.Warningf("Unable to get a peer ID from the pubkey, CID: %s: Error:%s\n", addr.String(), err.Error())	×
372	return	×
373	}	×
374
375	log.Debugf("Received offline message %s from: %s\n", addr.String(), id.Pretty())	×
376		×
377	if m.bm.IsBanned(id) {	×
378	log.Warningf("Received and dropped offline message from banned user: %s\n", id.Pretty())	×
379	return	×
380	}	×
381
382	if err := m.node.Peerstore.AddPubKey(id, pubkey); err != nil {	×
383	log.Errorf("adding pubkey to peerstore: %s", err.Error())	×
384	}	×
385	store := m.node.Repo.Datastore()	×
386	if err := ipfs.PutCachedPubkey(store, id.Pretty(), env.Pubkey); err != nil {	×
387	log.Errorf("caching pubkey: %s", err.Error())	×
388	}	×
389
390	// Respond with an ACK
391	if env.Message.MessageType != pb.Message_OFFLINE_ACK {	×
392	err = m.sendAck(id.Pretty(), pid)	×
393	if err != nil {	×
394	log.Error(err)	×
395	}	×
396	}
397
398	// handle
399	err = m.handleMessage(env, addr.String(), nil)	×
400	if err != nil {	×
401	log.Error(err)	×
402	}	×
403	}
404
405	// handleMessage loads the handler for this message type and attempts to process the message. Some message types (such
406	// as those partaining to an order) need to be processed in order. In these cases the handler returns a net.OutOfOrderMessage error
407	// and we must save the message to the database to await further processing.
408	func (m MessageRetriever) handleMessage(env pb.Envelope, addr string, id peer.ID) error {	×
409	if id == nil {	×
410	// Get the peer ID from the public key	×
411	pubkey, err := libp2p.UnmarshalPublicKey(env.Pubkey)	×
412	if err != nil {	×
413	log.Errorf("Error processing message %s. Type %s: %s", addr, env.Message.MessageType, err.Error())	×
414	return err	×
415	}	×
416	i, err := peer.IDFromPublicKey(pubkey)	×
417	if err != nil {	×
418	log.Errorf("Error processing message %s. Type %s: %s", addr, env.Message.MessageType, err.Error())	×
419	return err	×
420	}	×
421	id = &i	×
422	}
423
424	// Get handler for this message type
425	handler := m.service.HandlerForMsgType(env.Message.MessageType)	×
426	if handler == nil {	×
427	log.Errorf("Nil handler for message type %s", env.Message.MessageType)	×
428	return errors.New("nil handler for message")	×
429	}	×
430
431	// Dispatch handler
432	resp, err := handler(*id, env.Message, true)	×
433	if err != nil {	×
434	if err == net.OutOfOrderMessage {	×
435	ser, err := proto.Marshal(&env)	×
436	if err == nil {	×
437	err := m.db.OfflineMessages().SetMessage(addr, ser)	×
438	if err != nil {	×
439	log.Errorf("Error saving offline message %s to database: %s", addr, err.Error())	×
440	}	×
441	} else {	×
442	log.Errorf("Error serializing offline message %s for storage", addr)	×
443	}	×
444	} else if env.Message.MessageType == pb.Message_ORDER && resp != nil {	×
445	log.Errorf("Error processing ORDER message: %s, sending ERROR response", err.Error())	×
446	err = m.sendError(id.Pretty(), nil, *resp)	×
447	if err != nil {	×
448	log.Error(err)	×
449	}	×
450	return err	×
451	} else {	×
452	log.Errorf("Error processing message %s. Type %s: %s", addr, env.Message.MessageType, err.Error())	×
453	return err	×
454	}	×
455	}
456	return nil	×
457	}
458
459	var MessageProcessingOrder = []pb.Message_MessageType{
460	pb.Message_ORDER,
461	pb.Message_ORDER_CANCEL,
462	pb.Message_ORDER_REJECT,
463	pb.Message_ORDER_CONFIRMATION,
464	pb.Message_ORDER_PAYMENT,
465	pb.Message_ORDER_FULFILLMENT,
466	pb.Message_ORDER_COMPLETION,
467	pb.Message_DISPUTE_OPEN,
468	pb.Message_DISPUTE_UPDATE,
469	pb.Message_VENDOR_FINALIZED_PAYMENT,
470	pb.Message_DISPUTE_CLOSE,
471	pb.Message_REFUND,
472	pb.Message_CHAT,
473	pb.Message_FOLLOW,
474	pb.Message_UNFOLLOW,
475	pb.Message_MODERATOR_ADD,
476	pb.Message_MODERATOR_REMOVE,
477	pb.Message_OFFLINE_ACK,
478	pb.Message_OFFLINE_RELAY,
479	}
480
481	// processQueuedMessages loads all the saved messaged from the database for processing. For each message it sorts them into a
482	// queue based on message type and then processes the queue in order. Any messages that successfully process can then be deleted
483	// from the database.
484	func (m *MessageRetriever) processQueuedMessages() {	×
485	messageQueue := make(map[pb.Message_MessageType][]offlineMessage)	×
486	for _, messageType := range MessageProcessingOrder {	×
487	messageQueue[messageType] = []offlineMessage{}	×
488	}	×
489
490	// Load stored messages from database
491	messages, err := m.db.OfflineMessages().GetMessages()	×
492	if err != nil {	×
493	return	×
494	}	×
495	// Sort them into the queue by message type
496	for url, ser := range messages {	×
497	env := new(pb.Envelope)	×
498	err := proto.Unmarshal(ser, env)	×
499	if err == nil {	×
500	messageQueue[env.Message.MessageType] = append(messageQueue[env.Message.MessageType], offlineMessage{url, *env})	×
501	} else {	×
502	log.Error("Error unmarshalling serialized offline message from database")	×
503	}	×
504	}
505	var toDelete []string	×
506	// Process the queue in order	×
507	for _, messageType := range MessageProcessingOrder {	×
508	queue, ok := messageQueue[messageType]	×
509	if !ok {	×
510	continue	×
511	}
512	for _, om := range queue {	×
513	err := m.handleMessage(om.env, om.addr, nil)	×
514	if err == nil {	×
515	toDelete = append(toDelete, om.addr)	×
516	}	×
517	}
518	}
519	// Delete messages that we're successfully processed from the database
520	for _, url := range toDelete {	×
521	err = m.db.OfflineMessages().DeleteMessage(url)	×
522	if err != nil {	×
523	log.Error(err)	×
524	}	×
525	}
526	}

OpenBazaar / openbazaar-go / 7993

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous