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
use super::super::{Bitfield, Handshake, Message, WireItem};
use super::{CodecState, WireCodec};
use std::io;
use tokio_util::bytes::{self, Buf};
use tokio_util::codec::Decoder;
use tracing::debug;
impl Decoder for WireCodec {
type Item = WireItem;
type Error = std::io::Error;
fn decode(&mut self, src: &mut bytes::BytesMut) -> Result<Option<WireItem>, Self::Error> {
match &mut self.state {
CodecState::HandshakePending {
pstrlen,
info_hash_sent,
} => {
let mut offset = 0;
let handshake = {
let buf = src.as_ref();
// pstrlen
if src.len() < offset + 1 {
return Ok(None);
}
let len = if let Some(pstrlen) = pstrlen {
*pstrlen
} else {
*pstrlen = Some(buf[offset]);
buf[offset]
};
offset += 1;
// pstr
if src.len() < offset + len as usize {
return Ok(None);
}
// A peer controls these bytes, so they need not be UTF-8.
// Same answer as a pstr that decodes but says the wrong
// thing: this is not a BitTorrent peer.
let Ok(pstr) = std::str::from_utf8(&buf[1..=len as usize]) else {
return Err(std::io::Error::new(
io::ErrorKind::InvalidData,
"invalid pstr",
));
};
if pstr != "BitTorrent protocol" {
return Err(std::io::Error::new(
io::ErrorKind::InvalidData,
"invalid pstr",
));
}
offset += len as usize;
// reserved
if src.len() < offset + 8 {
return Ok(None);
}
let reserved: [u8; 8] = buf[offset..offset + 8]
.try_into()
.expect("failed to copy reserved");
offset += 8;
// info_hash
if src.len() < offset + 20 {
return Ok(None);
}
if !*info_hash_sent {
let info_hash = buf[offset..offset + 20]
.try_into()
.expect("failed to copy info_hash");
*info_hash_sent = true;
return Ok(Some(WireItem::HandshakePartial { info_hash }));
}
let info_hash: [u8; 20] = buf[offset..offset + 20]
.try_into()
.expect("failed to copy info_hash");
offset += 20;
// peer_id
if src.len() < offset + 20 {
return Ok(None);
}
let peer_id: [u8; 20] = buf[offset..offset + 20]
.try_into()
.expect("failed to copy peer_id");
offset += 20;
Handshake {
pstrlen: len,
pstr: pstr.to_string(),
info_hash,
reserved,
peer_id,
}
};
src.advance(offset);
self.state = CodecState::Normal;
Ok(Some(WireItem::Handshake(handshake)))
}
CodecState::Normal => loop {
if src.len() < 4 {
return Ok(None);
}
let msg_len = u32::from_be_bytes(src[..4].try_into().unwrap()) as usize;
if src.len() < 4 + msg_len {
return Ok(None);
}
// Keep-alive: no id, no payload. It is passed on rather than
// swallowed, because the idle timeout upstream is counting
// silence and this is the one message that breaks it.
if msg_len == 0 {
src.advance(4);
return Ok(Some(WireItem::Message(Message::KeepAlive)));
}
let payload = &src[4..4 + msg_len];
let id = payload[0];
let rest = &payload[1..];
let msg = match id {
0 => Message::Choke,
1 => Message::Unchoke,
2 => Message::Interested,
3 => Message::NotInterested,
4 => {
if rest.len() < 4 {
return Err(io::Error::new(io::ErrorKind::InvalidData, "Invalid Have"));
}
Message::Have(u32::from_be_bytes(rest[..4].try_into().unwrap()))
}
5 => Message::Bitfield(Bitfield(rest.to_vec())),
6 => {
if rest.len() < 12 {
return Err(io::Error::new(
io::ErrorKind::InvalidData,
"Invalid Request",
));
}
let index = u32::from_be_bytes(rest[0..4].try_into().unwrap());
let begin = u32::from_be_bytes(rest[4..8].try_into().unwrap());
let length = u32::from_be_bytes(rest[8..12].try_into().unwrap());
Message::Request {
index,
begin,
length,
}
}
7 => {
if rest.len() < 8 {
return Err(io::Error::new(
io::ErrorKind::InvalidData,
"Invalid Piece",
));
}
let index = u32::from_be_bytes(rest[0..4].try_into().unwrap());
let begin = u32::from_be_bytes(rest[4..8].try_into().unwrap());
let block = rest[8..].to_vec();
Message::Piece {
index,
begin,
block,
}
}
8 => {
if rest.len() < 12 {
return Err(io::Error::new(
io::ErrorKind::InvalidData,
"Invalid Cancel",
));
}
let index = u32::from_be_bytes(rest[0..4].try_into().unwrap());
let begin = u32::from_be_bytes(rest[4..8].try_into().unwrap());
let length = u32::from_be_bytes(rest[8..12].try_into().unwrap());
Message::Cancel {
index,
begin,
length,
}
}
9 => {
if rest.len() < 2 {
return Err(io::Error::new(io::ErrorKind::InvalidData, "Invalid Port"));
}
Message::Port(u16::from_be_bytes(rest[..2].try_into().unwrap()))
}
_ => {
// Unknown id, most likely an extension we never
// advertised. Frames are length-prefixed, so skip this
// one and keep the connection alive.
debug!("skipping unknown message id {}", id);
src.advance(4 + msg_len);
continue;
}
};
src.advance(4 + msg_len);
return Ok(Some(WireItem::Message(msg)));
},
}
}
}