aboutsummaryrefslogtreecommitdiffstats
path: root/p2p/protocol.go
blob: 969937076bd68f09d9231e0585db7c586a518e19 (plain) (blame)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
package p2p

import (
    "bytes"
    "time"
)

// Protocol represents a P2P subprotocol implementation.
type Protocol struct {
    // Name should contain the official protocol name,
    // often a three-letter word.
    Name string

    // Version should contain the version number of the protocol.
    Version uint

    // Length should contain the number of message codes used
    // by the protocol.
    Length uint64

    // Run is called in a new groutine when the protocol has been
    // negotiated with a peer. It should read and write messages from
    // rw. The Payload for each message must be fully consumed.
    //
    // The peer connection is closed when Start returns. It should return
    // any protocol-level error (such as an I/O error) that is
    // encountered.
    Run func(peer *Peer, rw MsgReadWriter) error
}

func (p Protocol) cap() Cap {
    return Cap{p.Name, p.Version}
}

const (
    baseProtocolVersion    = 2
    baseProtocolLength     = uint64(16)
    baseProtocolMaxMsgSize = 10 * 1024 * 1024
)

const (
    // devp2p message codes
    handshakeMsg = 0x00
    discMsg      = 0x01
    pingMsg      = 0x02
    pongMsg      = 0x03
    getPeersMsg  = 0x04
    peersMsg     = 0x05
)

// handshake is the structure of a handshake list.
type handshake struct {
    Version    uint64
    ID         string
    Caps       []Cap
    ListenPort uint64
    NodeID     []byte
}

func (h *handshake) String() string {
    return h.ID
}
func (h *handshake) Pubkey() []byte {
    return h.NodeID
}

// Cap is the structure of a peer capability.
type Cap struct {
    Name    string
    Version uint
}

func (cap Cap) RlpData() interface{} {
    return []interface{}{cap.Name, cap.Version}
}

type capsByName []Cap

func (cs capsByName) Len() int           { return len(cs) }
func (cs capsByName) Less(i, j int) bool { return cs[i].Name < cs[j].Name }
func (cs capsByName) Swap(i, j int)      { cs[i], cs[j] = cs[j], cs[i] }

type baseProtocol struct {
    rw   MsgReadWriter
    peer *Peer
}

func runBaseProtocol(peer *Peer, rw MsgReadWriter) error {
    bp := &baseProtocol{rw, peer}
    errc := make(chan error, 1)
    go func() { errc <- rw.WriteMsg(bp.handshakeMsg()) }()
    if err := bp.readHandshake(); err != nil {
        return err
    }
    // handle write error
    if err := <-errc; err != nil {
        return err
    }
    // run main loop
    go func() {
        for {
            if err := bp.handle(rw); err != nil {
                errc <- err
                break
            }
        }
    }()
    return bp.loop(errc)
}

var pingTimeout = 2 * time.Second

func (bp *baseProtocol) loop(quit <-chan error) error {
    ping := time.NewTimer(pingTimeout)
    activity := bp.peer.activity.Subscribe(time.Time{})
    lastActive := time.Time{}
    defer ping.Stop()
    defer activity.Unsubscribe()

    getPeersTick := time.NewTicker(10 * time.Second)
    defer getPeersTick.Stop()
    err := EncodeMsg(bp.rw, getPeersMsg)

    for err == nil {
        select {
        case err = <-quit:
            return err
        case <-getPeersTick.C:
            err = EncodeMsg(bp.rw, getPeersMsg)
        case event := <-activity.Chan():
            ping.Reset(pingTimeout)
            lastActive = event.(time.Time)
        case t := <-ping.C:
            if lastActive.Add(pingTimeout * 2).Before(t) {
                err = newPeerError(errPingTimeout, "")
            } else if lastActive.Add(pingTimeout).Before(t) {
                err = EncodeMsg(bp.rw, pingMsg)
            }
        }
    }
    return err
}

func (bp *baseProtocol) handle(rw MsgReadWriter) error {
    msg, err := rw.ReadMsg()
    if err != nil {
        return err
    }
    if msg.Size > baseProtocolMaxMsgSize {
        return newPeerError(errMisc, "message too big")
    }
    // make sure that the payload has been fully consumed
    defer msg.Discard()

    switch msg.Code {
    case handshakeMsg:
        return newPeerError(errProtocolBreach, "extra handshake received")

    case discMsg:
        var reason [1]DiscReason
        if err := msg.Decode(&reason); err != nil {
            return err
        }
        return discRequestedError(reason[0])

    case pingMsg:
        return EncodeMsg(bp.rw, pongMsg)

    case pongMsg:

    case getPeersMsg:
        peers := bp.peer.PeerList()
        // this is dangerous. the spec says that we should _delay_
        // sending the response if no new information is available.
        // this means that would need to send a response later when
        // new peers become available.
        //
        // TODO: add event mechanism to notify baseProtocol for new peers
        if len(peers) > 0 {
            return EncodeMsg(bp.rw, peersMsg, peers...)
        }

    case peersMsg:
        var peers []*peerAddr
        if err := msg.Decode(&peers); err != nil {
            return err
        }
        for _, addr := range peers {
            bp.peer.Debugf("received peer suggestion: %v", addr)
            bp.peer.newPeerAddr <- addr
        }

    default:
        return newPeerError(errInvalidMsgCode, "unknown message code %v", msg.Code)
    }
    return nil
}

func (bp *baseProtocol) readHandshake() error {
    // read and handle remote handshake
    msg, err := bp.rw.ReadMsg()
    if err != nil {
        return err
    }
    if msg.Code != handshakeMsg {
        return newPeerError(errProtocolBreach, "first message must be handshake, got %x", msg.Code)
    }
    if msg.Size > baseProtocolMaxMsgSize {
        return newPeerError(errMisc, "message too big")
    }
    var hs handshake
    if err := msg.Decode(&hs); err != nil {
        return err
    }
    // validate handshake info
    if hs.Version != baseProtocolVersion {
        return newPeerError(errP2PVersionMismatch, "Require protocol %d, received %d\n",
            baseProtocolVersion, hs.Version)
    }
    if len(hs.NodeID) == 0 {
        return newPeerError(errPubkeyMissing, "")
    }
    if len(hs.NodeID) != 64 {
        return newPeerError(errPubkeyInvalid, "require 512 bit, got %v", len(hs.NodeID)*8)
    }
    if da := bp.peer.dialAddr; da != nil {
        // verify that the peer we wanted to connect to
        // actually holds the target public key.
        if da.Pubkey != nil && !bytes.Equal(da.Pubkey, hs.NodeID) {
            return newPeerError(errPubkeyForbidden, "dial address pubkey mismatch")
        }
    }
    pa := newPeerAddr(bp.peer.conn.RemoteAddr(), hs.NodeID)
    if err := bp.peer.pubkeyHook(pa); err != nil {
        return newPeerError(errPubkeyForbidden, "%v", err)
    }
    // TODO: remove Caps with empty name
    var addr *peerAddr
    if hs.ListenPort != 0 {
        addr = newPeerAddr(bp.peer.conn.RemoteAddr(), hs.NodeID)
        addr.Port = hs.ListenPort
    }
    bp.peer.setHandshakeInfo(&hs, addr, hs.Caps)
    bp.peer.startSubprotocols(hs.Caps)
    return nil
}

func (bp *baseProtocol) handshakeMsg() Msg {
    var (
        port uint64
        caps []interface{}
    )
    if bp.peer.ourListenAddr != nil {
        port = bp.peer.ourListenAddr.Port
    }
    for _, proto := range bp.peer.protocols {
        caps = append(caps, proto.cap())
    }
    return NewMsg(handshakeMsg,
        baseProtocolVersion,
        bp.peer.ourID.String(),
        caps,
        port,
        bp.peer.ourID.Pubkey()[1:],
    )
}