peer.go raw
1 // Package p2p is a minimal Bitcoin P2P peer: the version handshake, frame
2 // reading and writing, ping/pong, and the message loop the compact-filter
3 // client drives. It knows nothing about wallets or chain state.
4 package p2p
5
6 import (
7 "context"
8 "crypto/rand"
9 "encoding/binary"
10 "errors"
11 "fmt"
12 "net"
13 "sync"
14 "time"
15
16 "git.smesh.lol/nostrino/pkg/btcwire"
17 )
18
19 // DefaultUserAgent identifies this client to a chain node.
20 const DefaultUserAgent = "/nostrino:0.1.0/"
21
22 // Config configures one peer connection.
23 type Config struct {
24 // Magic is the chain's P2P network magic.
25 Magic uint32
26 // UserAgent is the BIP14 user agent. Empty means DefaultUserAgent.
27 UserAgent string
28 // Services is the service bits this client advertises. A light client
29 // advertises none by default.
30 Services uint64
31 // LastBlock is the caller's best height, sent in version.
32 LastBlock int32
33 // ReadTimeout bounds one frame read. Zero means 60s.
34 ReadTimeout time.Duration
35 // RequireCompactFilters rejects a peer that does not advertise NODE
36 // COMPACT_FILTERS.
37 RequireCompactFilters bool
38 // Logf receives diagnostics; nil discards them.
39 Logf func(string, ...any)
40 }
41
42 // Conn is one P2P connection.
43 type Conn struct {
44 conn net.Conn
45 reader *btcwire.Reader
46 cfg Config
47
48 writeMu sync.Mutex
49 closeMu sync.Mutex
50 closed bool
51
52 peer *btcwire.Version
53 }
54
55 // New wraps an established byte stream (a TCP connection or a tunnel Conn).
56 func New(conn net.Conn, cfg Config) *Conn {
57 if cfg.Magic == 0 {
58 panic("p2p: zero network magic")
59 }
60 if cfg.ReadTimeout == 0 {
61 cfg.ReadTimeout = 60 * time.Second
62 }
63 if cfg.UserAgent == "" {
64 cfg.UserAgent = DefaultUserAgent
65 }
66 if cfg.Logf == nil {
67 cfg.Logf = func(string, ...any) {}
68 }
69 return &Conn{
70 conn: conn,
71 reader: btcwire.NewReader(conn, cfg.Magic),
72 cfg: cfg,
73 }
74 }
75
76 // DialTCP opens a TCP connection and wraps it.
77 func DialTCP(ctx context.Context, addr string, cfg Config) (*Conn, error) {
78 var d net.Dialer
79 conn, err := d.DialContext(ctx, "tcp", addr)
80 if err != nil {
81 return nil, fmt.Errorf("p2p: dial %s: %w", addr, err)
82 }
83 return New(conn, cfg), nil
84 }
85
86 // PeerVersion is the version message received from the remote peer.
87 func (c *Conn) PeerVersion() *btcwire.Version { return c.peer }
88
89 // Handshake performs version/verack. It returns the peer's version message.
90 func (c *Conn) Handshake(ctx context.Context) (*btcwire.Version, error) {
91 our := &btcwire.Version{
92 ProtocolVersion: btcwire.ProtocolVersion,
93 Services: c.cfg.Services,
94 Timestamp: time.Now().Unix(),
95 Nonce: randomNonce(),
96 UserAgent: c.cfg.UserAgent,
97 LastBlock: c.cfg.LastBlock,
98 Relay: false,
99 }
100 if err := c.Send(btcwire.CmdVersion, our); err != nil {
101 return nil, err
102 }
103
104 var peer *btcwire.Version
105 for peer == nil {
106 msg, err := c.Next(ctx)
107 if err != nil {
108 return nil, fmt.Errorf("p2p: waiting for version: %w", err)
109 }
110 if msg.Command != btcwire.CmdVersion {
111 continue
112 }
113 var v btcwire.Version
114 if err := v.DecodePayload(msg.Payload); err != nil {
115 return nil, fmt.Errorf("p2p: decode version: %w", err)
116 }
117 peer = &v
118 }
119 if peer.ProtocolVersion < 70015 {
120 return nil, fmt.Errorf("p2p: peer protocol version %d is too old for compact filters", peer.ProtocolVersion)
121 }
122 if c.cfg.RequireCompactFilters && peer.Services&btcwire.ServiceNodeCompactFilters == 0 {
123 return nil, fmt.Errorf("p2p: peer does not serve compact filters (services %#x)", peer.Services)
124 }
125 if err := c.SendEmpty(btcwire.CmdVerAck); err != nil {
126 return nil, err
127 }
128 // A node may announce sendheaders/wtxidrelay before verack.
129 for {
130 msg, err := c.Next(ctx)
131 if err != nil {
132 return nil, fmt.Errorf("p2p: waiting for verack: %w", err)
133 }
134 if msg.Command == btcwire.CmdVerAck {
135 break
136 }
137 }
138 c.peer = peer
139 if err := c.SendEmpty(btcwire.CmdSendHeaders); err != nil {
140 return nil, err
141 }
142 return peer, nil
143 }
144
145 func randomNonce() uint64 {
146 var b [8]byte
147 if _, err := rand.Read(b[:]); err != nil {
148 return uint64(time.Now().UnixNano())
149 }
150 return binary.LittleEndian.Uint64(b[:])
151 }
152
153 // Send encodes and writes one message.
154 func (c *Conn) Send(command string, payload btcwire.Payload) error {
155 var raw []byte
156 if payload != nil {
157 b, err := payload.EncodePayload()
158 if err != nil {
159 return err
160 }
161 raw = b
162 }
163 frame, err := btcwire.EncodeMessage(c.cfg.Magic, command, raw)
164 if err != nil {
165 return err
166 }
167 c.writeMu.Lock()
168 defer c.writeMu.Unlock()
169 c.closeMu.Lock()
170 closed := c.closed
171 c.closeMu.Unlock()
172 if closed {
173 return errors.New("p2p: closed")
174 }
175 if _, err := c.conn.Write(frame); err != nil {
176 return fmt.Errorf("p2p: write %s: %w", command, err)
177 }
178 return nil
179 }
180
181 // SendEmpty writes a command with an empty payload.
182 func (c *Conn) SendEmpty(command string) error { return c.Send(command, nil) }
183
184 // Next reads the next frame. Ping is answered automatically and not returned.
185 func (c *Conn) Next(ctx context.Context) (*btcwire.Message, error) {
186 for {
187 if err := c.setReadDeadline(ctx); err != nil {
188 return nil, err
189 }
190 msg, err := c.reader.Next()
191 if err != nil {
192 return nil, err
193 }
194 if msg.Command == btcwire.CmdPing {
195 var p btcwire.Ping
196 if err := p.DecodePayload(msg.Payload); err != nil {
197 return nil, fmt.Errorf("p2p: decode ping: %w", err)
198 }
199 if err := c.Send(btcwire.CmdPong, &btcwire.Pong{Nonce: p.Nonce}); err != nil {
200 return nil, err
201 }
202 continue
203 }
204 return msg, nil
205 }
206 }
207
208 func (c *Conn) setReadDeadline(ctx context.Context) error {
209 deadline := time.Now().Add(c.cfg.ReadTimeout)
210 if d, ok := ctx.Deadline(); ok && d.Before(deadline) {
211 deadline = d
212 }
213 if err := c.conn.SetReadDeadline(deadline); err != nil {
214 return err
215 }
216 select {
217 case <-ctx.Done():
218 return ctx.Err()
219 default:
220 return nil
221 }
222 }
223
224 // ReadUntil reads frames until one with the given command arrives, returning
225 // it. Other messages are discarded.
226 func (c *Conn) ReadUntil(ctx context.Context, command string) (*btcwire.Message, error) {
227 for {
228 msg, err := c.Next(ctx)
229 if err != nil {
230 return nil, err
231 }
232 if msg.Command == command {
233 return msg, nil
234 }
235 }
236 }
237
238 // ReadUntilMax reads frames until one with the given command arrives, refusing any frame whose
239 // payload exceeds max before the payload is allocated.
240 //
241 // It exists for one reason: a peer answering getdata can answer with anything, and a client that
242 // allocates whatever the header claims has handed a stranger its memory. The limit is per call
243 // and restored afterwards, so a block read cannot loosen the cap every other read lives under.
244 func (c *Conn) ReadUntilMax(ctx context.Context, command string, max int) (*btcwire.Message, error) {
245 c.reader.SetLimit(max)
246
247 defer c.reader.SetLimit(btcwire.MaxPayload)
248
249 for {
250 msg, err := c.Next(ctx)
251 if err != nil {
252 return nil, err
253 }
254
255 if msg.Command == command {
256 return msg, nil
257 }
258 }
259 }
260
261 // Close closes the underlying connection.
262 func (c *Conn) Close() error {
263 c.closeMu.Lock()
264 if c.closed {
265 c.closeMu.Unlock()
266 return nil
267 }
268 c.closed = true
269 c.closeMu.Unlock()
270 return c.conn.Close()
271 }
272
273 // Conn exposes the underlying stream for callers that need deadlines.
274 func (c *Conn) Conn() net.Conn { return c.conn }
275