tcp.rs 32 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908
  1. use core::fmt;
  2. use Error;
  3. use Managed;
  4. use wire::{IpProtocol, IpAddress, IpEndpoint};
  5. use wire::{TcpPacket, TcpRepr, TcpControl};
  6. use socket::{Socket, IpRepr, IpPayload};
  7. /// A TCP stream ring buffer.
  8. #[derive(Debug)]
  9. pub struct SocketBuffer<'a> {
  10. storage: Managed<'a, [u8]>,
  11. read_at: usize,
  12. length: usize
  13. }
  14. impl<'a> SocketBuffer<'a> {
  15. /// Create a packet buffer with the given storage.
  16. pub fn new<T>(storage: T) -> SocketBuffer<'a>
  17. where T: Into<Managed<'a, [u8]>> {
  18. SocketBuffer {
  19. storage: storage.into(),
  20. read_at: 0,
  21. length: 0
  22. }
  23. }
  24. fn capacity(&self) -> usize {
  25. self.storage.len()
  26. }
  27. fn len(&self) -> usize {
  28. self.length
  29. }
  30. fn window(&self) -> usize {
  31. self.capacity() - self.len()
  32. }
  33. fn clamp_writer(&self, mut size: usize) -> (usize, usize) {
  34. let write_at = (self.read_at + self.length) % self.storage.len();
  35. // We can't enqueue more than there is free space.
  36. let free = self.storage.len() - self.length;
  37. if size > free { size = free }
  38. // We can't contiguously enqueue past the beginning of the storage.
  39. let until_end = self.storage.len() - write_at;
  40. if size > until_end { size = until_end }
  41. (write_at, size)
  42. }
  43. fn enqueue(&mut self, size: usize) -> &mut [u8] {
  44. let (write_at, size) = self.clamp_writer(size);
  45. self.length += size;
  46. &mut self.storage[write_at..write_at + size]
  47. }
  48. fn enqueue_slice(&mut self, data: &[u8]) {
  49. let data = {
  50. let mut dest = self.enqueue(data.len());
  51. let (data, rest) = data.split_at(dest.len());
  52. dest.copy_from_slice(data);
  53. rest
  54. };
  55. // Retry, in case we had a wraparound.
  56. let mut dest = self.enqueue(data.len());
  57. let (data, _) = data.split_at(dest.len());
  58. dest.copy_from_slice(data);
  59. }
  60. fn clamp_reader(&self, mut size: usize) -> (usize, usize) {
  61. let read_at = self.read_at;
  62. // We can't dequeue more than was queued.
  63. if size > self.length { size = self.length }
  64. // We can't contiguously dequeue past the end of the storage.
  65. let until_end = self.storage.len() - read_at;
  66. if size > until_end { size = until_end }
  67. (read_at, size)
  68. }
  69. fn peek(&self, size: usize) -> &[u8] {
  70. let (read_at, size) = self.clamp_reader(size);
  71. &self.storage[read_at..read_at + size]
  72. }
  73. fn advance(&mut self, size: usize) {
  74. let (read_at, size) = self.clamp_reader(size);
  75. self.read_at = (read_at + size) % self.storage.len();
  76. self.length -= size;
  77. }
  78. #[allow(dead_code)] // only used in tests
  79. fn dequeue(&mut self, size: usize) -> &[u8] {
  80. let (read_at, size) = self.clamp_reader(size);
  81. self.read_at = (self.read_at + size) % self.storage.len();
  82. self.length -= size;
  83. &self.storage[read_at..read_at + size]
  84. }
  85. }
  86. impl<'a> Into<SocketBuffer<'a>> for Managed<'a, [u8]> {
  87. fn into(self) -> SocketBuffer<'a> {
  88. SocketBuffer::new(self)
  89. }
  90. }
  91. #[derive(Debug, PartialEq, Eq, Clone, Copy)]
  92. pub enum State {
  93. Closed,
  94. Listen,
  95. SynSent,
  96. SynReceived,
  97. Established,
  98. FinWait1,
  99. FinWait2,
  100. CloseWait,
  101. Closing,
  102. LastAck,
  103. TimeWait
  104. }
  105. impl fmt::Display for State {
  106. fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
  107. match self {
  108. &State::Closed => write!(f, "CLOSED"),
  109. &State::Listen => write!(f, "LISTEN"),
  110. &State::SynSent => write!(f, "SYN_SENT"),
  111. &State::SynReceived => write!(f, "SYN_RECEIVED"),
  112. &State::Established => write!(f, "ESTABLISHED"),
  113. &State::FinWait1 => write!(f, "FIN_WAIT_1"),
  114. &State::FinWait2 => write!(f, "FIN_WAIT_2"),
  115. &State::CloseWait => write!(f, "CLOSE_WAIT"),
  116. &State::Closing => write!(f, "CLOSING"),
  117. &State::LastAck => write!(f, "LAST_ACK"),
  118. &State::TimeWait => write!(f, "TIME_WAIT")
  119. }
  120. }
  121. }
  122. #[derive(Debug)]
  123. struct Retransmit {
  124. sent: bool // FIXME
  125. }
  126. impl Retransmit {
  127. fn new() -> Retransmit {
  128. Retransmit { sent: false }
  129. }
  130. fn reset(&mut self) {
  131. self.sent = false
  132. }
  133. fn check(&mut self) -> bool {
  134. let result = !self.sent;
  135. self.sent = true;
  136. result
  137. }
  138. }
  139. /// A Transmission Control Protocol data stream.
  140. #[derive(Debug)]
  141. pub struct TcpSocket<'a> {
  142. state: State,
  143. listen_address: IpAddress,
  144. local_endpoint: IpEndpoint,
  145. remote_endpoint: IpEndpoint,
  146. local_seq_no: i32,
  147. remote_seq_no: i32,
  148. remote_last_ack: i32,
  149. remote_win_len: usize,
  150. retransmit: Retransmit,
  151. rx_buffer: SocketBuffer<'a>,
  152. tx_buffer: SocketBuffer<'a>
  153. }
  154. impl<'a> TcpSocket<'a> {
  155. /// Create a socket using the given buffers.
  156. pub fn new<T>(rx_buffer: T, tx_buffer: T) -> Socket<'a, 'static>
  157. where T: Into<SocketBuffer<'a>> {
  158. let rx_buffer = rx_buffer.into();
  159. if rx_buffer.capacity() > <u16>::max_value() as usize {
  160. panic!("buffers larger than {} require window scaling, which is not implemented",
  161. <u16>::max_value())
  162. }
  163. Socket::Tcp(TcpSocket {
  164. state: State::Closed,
  165. listen_address: IpAddress::default(),
  166. local_endpoint: IpEndpoint::default(),
  167. remote_endpoint: IpEndpoint::default(),
  168. local_seq_no: 0,
  169. remote_seq_no: 0,
  170. remote_win_len: 0,
  171. remote_last_ack: 0,
  172. retransmit: Retransmit::new(),
  173. tx_buffer: tx_buffer.into(),
  174. rx_buffer: rx_buffer.into()
  175. })
  176. }
  177. /// Return the connection state.
  178. #[inline(always)]
  179. pub fn state(&self) -> State {
  180. self.state
  181. }
  182. /// Return the local endpoint.
  183. #[inline(always)]
  184. pub fn local_endpoint(&self) -> IpEndpoint {
  185. self.local_endpoint
  186. }
  187. /// Return the remote endpoint.
  188. #[inline(always)]
  189. pub fn remote_endpoint(&self) -> IpEndpoint {
  190. self.remote_endpoint
  191. }
  192. fn set_state(&mut self, state: State) {
  193. if self.state != state {
  194. if self.remote_endpoint.addr.is_unspecified() {
  195. net_trace!("tcp:{}: state={}→{}",
  196. self.local_endpoint, self.state, state);
  197. } else {
  198. net_trace!("tcp:{}:{}: state={}→{}",
  199. self.local_endpoint, self.remote_endpoint, self.state, state);
  200. }
  201. }
  202. self.state = state
  203. }
  204. /// Start listening on the given endpoint.
  205. ///
  206. /// # Panics
  207. /// This function will panic if the socket is not in the CLOSED state.
  208. pub fn listen(&mut self, endpoint: IpEndpoint) {
  209. assert!(self.state == State::Closed);
  210. self.listen_address = endpoint.addr;
  211. self.local_endpoint = endpoint;
  212. self.remote_endpoint = IpEndpoint::default();
  213. self.set_state(State::Listen);
  214. }
  215. /// See [Socket::collect](enum.Socket.html#method.collect).
  216. pub fn collect(&mut self, ip_repr: &IpRepr, payload: &[u8]) -> Result<(), Error> {
  217. if ip_repr.protocol() != IpProtocol::Tcp { return Err(Error::Rejected) }
  218. let packet = try!(TcpPacket::new(payload));
  219. let repr = try!(TcpRepr::parse(&packet, &ip_repr.src_addr(), &ip_repr.dst_addr()));
  220. // Reject packets with a wrong destination.
  221. if self.local_endpoint.port != repr.dst_port { return Err(Error::Rejected) }
  222. if !self.local_endpoint.addr.is_unspecified() &&
  223. self.local_endpoint.addr != ip_repr.dst_addr() { return Err(Error::Rejected) }
  224. // Reject packets from a source to which we aren't connected.
  225. if self.remote_endpoint.port != 0 &&
  226. self.remote_endpoint.port != repr.src_port { return Err(Error::Rejected) }
  227. if !self.remote_endpoint.addr.is_unspecified() &&
  228. self.remote_endpoint.addr != ip_repr.src_addr() { return Err(Error::Rejected) }
  229. // Reject packets addressed to a closed socket.
  230. if self.state == State::Closed {
  231. net_trace!("tcp:{}:{}:{}: packet received by a closed socket",
  232. self.local_endpoint, ip_repr.src_addr(), repr.src_port);
  233. return Err(Error::Malformed)
  234. }
  235. // Reject unacceptable acknowledgements.
  236. match (self.state, repr) {
  237. // The initial SYN (or whatever) cannot contain an acknowledgement.
  238. (State::Listen, TcpRepr { ack_number: Some(_), .. }) => {
  239. net_trace!("tcp:{}:{}: ACK received by a socket in LISTEN state",
  240. self.local_endpoint, self.remote_endpoint);
  241. return Err(Error::Malformed)
  242. }
  243. (State::Listen, TcpRepr { ack_number: None, .. }) => (),
  244. // A reset received in response to initial SYN is acceptable if it acknowledges
  245. // the initial SYN.
  246. (State::SynSent, TcpRepr { control: TcpControl::Rst, ack_number: None, .. }) => {
  247. net_trace!("tcp:{}:{}: unacceptable RST (expecting RST|ACK) \
  248. in response to initial SYN",
  249. self.local_endpoint, self.remote_endpoint);
  250. return Err(Error::Malformed)
  251. }
  252. (State::SynSent, TcpRepr {
  253. control: TcpControl::Rst, ack_number: Some(ack_number), ..
  254. }) => {
  255. if ack_number != self.local_seq_no {
  256. net_trace!("tcp:{}:{}: unacceptable RST|ACK in response to initial SYN",
  257. self.local_endpoint, self.remote_endpoint);
  258. return Err(Error::Malformed)
  259. }
  260. }
  261. // Every packet after the initial SYN must be an acknowledgement.
  262. (_, TcpRepr { ack_number: None, .. }) => {
  263. net_trace!("tcp:{}:{}: expecting an ACK",
  264. self.local_endpoint, self.remote_endpoint);
  265. return Err(Error::Malformed)
  266. }
  267. // Every acknowledgement must be for transmitted but unacknowledged data.
  268. (state, TcpRepr { ack_number: Some(ack_number), .. }) => {
  269. let control_len =
  270. if state == State::SynReceived { 1 } else { 0 };
  271. let unacknowledged = self.tx_buffer.len() as i32 + control_len;
  272. if !(ack_number - self.local_seq_no >= 0 &&
  273. ack_number - (self.local_seq_no + unacknowledged) <= 0) {
  274. net_trace!("tcp:{}:{}: unacceptable ACK ({} not in {}..{})",
  275. self.local_endpoint, self.remote_endpoint,
  276. ack_number, self.local_seq_no, self.local_seq_no + unacknowledged);
  277. return Err(Error::Malformed)
  278. }
  279. }
  280. }
  281. match (self.state, repr) {
  282. // In LISTEN and SYN_SENT states, we have not yet synchronized with the remote end.
  283. (State::Listen, _) => (),
  284. (State::SynSent, _) => (),
  285. // In all other states, segments must occupy a valid portion of the receive window.
  286. // For now, do not try to reassemble out-of-order segments.
  287. (_, TcpRepr { control, seq_number, .. }) => {
  288. let next_remote_seq = self.remote_seq_no + self.rx_buffer.len() as i32 +
  289. control.len();
  290. if seq_number - next_remote_seq > 0 {
  291. net_trace!("tcp:{}:{}: unacceptable SEQ ({} not in {}..)",
  292. self.local_endpoint, self.remote_endpoint,
  293. seq_number, next_remote_seq);
  294. return Err(Error::Malformed)
  295. } else if seq_number - next_remote_seq != 0 {
  296. net_trace!("tcp:{}:{}: duplicate SEQ ({} in ..{})",
  297. self.local_endpoint, self.remote_endpoint,
  298. seq_number, next_remote_seq);
  299. return Ok(())
  300. }
  301. }
  302. }
  303. // Validate and update the state.
  304. let old_state = self.state;
  305. match (self.state, repr) {
  306. // RSTs are ignored in the LISTEN state.
  307. (State::Listen, TcpRepr { control: TcpControl::Rst, .. }) =>
  308. return Ok(()),
  309. // RSTs in SYN_RECEIVED flip the socket back to the LISTEN state.
  310. (State::SynReceived, TcpRepr { control: TcpControl::Rst, .. }) => {
  311. self.local_endpoint.addr = self.listen_address;
  312. self.remote_endpoint = IpEndpoint::default();
  313. self.set_state(State::Listen);
  314. return Ok(())
  315. }
  316. // RSTs in any other state close the socket.
  317. (_, TcpRepr { control: TcpControl::Rst, .. }) => {
  318. self.local_endpoint = IpEndpoint::default();
  319. self.remote_endpoint = IpEndpoint::default();
  320. self.set_state(State::Closed);
  321. return Ok(())
  322. }
  323. // SYN packets in the LISTEN state change it to SYN_RECEIVED.
  324. (State::Listen, TcpRepr {
  325. src_port, dst_port, control: TcpControl::Syn, seq_number, ack_number: None, ..
  326. }) => {
  327. self.local_endpoint = IpEndpoint::new(ip_repr.dst_addr(), dst_port);
  328. self.remote_endpoint = IpEndpoint::new(ip_repr.src_addr(), src_port);
  329. self.local_seq_no = -seq_number; // FIXME: use something more secure
  330. self.remote_seq_no = seq_number + 1;
  331. self.set_state(State::SynReceived);
  332. self.retransmit.reset()
  333. }
  334. // SYN|ACK packets in the SYN_RECEIVED state change it to ESTABLISHED.
  335. (State::SynReceived, TcpRepr { control: TcpControl::None, .. }) => {
  336. self.set_state(State::Established);
  337. self.retransmit.reset()
  338. }
  339. // ACK packets in ESTABLISHED state do nothing.
  340. (State::Established, TcpRepr { control: TcpControl::None, .. }) => (),
  341. _ => {
  342. net_trace!("tcp:{}:{}: unexpected packet {}",
  343. self.local_endpoint, self.remote_endpoint, repr);
  344. return Err(Error::Malformed)
  345. }
  346. }
  347. // Dequeue acknowledged octets.
  348. if let Some(ack_number) = repr.ack_number {
  349. let control_len =
  350. if old_state == State::SynReceived { 1 } else { 0 };
  351. if control_len > 0 {
  352. net_trace!("tcp:{}:{}: ACK for a control flag",
  353. self.local_endpoint, self.remote_endpoint);
  354. }
  355. if ack_number - self.local_seq_no - control_len > 0 {
  356. net_trace!("tcp:{}:{}: ACK for {} octets",
  357. self.local_endpoint, self.remote_endpoint,
  358. ack_number - self.local_seq_no - control_len);
  359. }
  360. self.tx_buffer.advance((ack_number - self.local_seq_no - control_len) as usize);
  361. self.local_seq_no = ack_number;
  362. }
  363. // Enqueue payload octets, which is guaranteed to be in order, unless we already did.
  364. if repr.payload.len() > 0 {
  365. net_trace!("tcp:{}:{}: receiving {} octets",
  366. self.local_endpoint, self.remote_endpoint, repr.payload.len());
  367. self.rx_buffer.enqueue_slice(repr.payload)
  368. }
  369. // Update window length.
  370. self.remote_win_len = repr.window_len as usize;
  371. Ok(())
  372. }
  373. /// See [Socket::dispatch](enum.Socket.html#method.dispatch).
  374. pub fn dispatch<F, R>(&mut self, emit: &mut F) -> Result<R, Error>
  375. where F: FnMut(&IpRepr, &IpPayload) -> Result<R, Error> {
  376. let ip_repr = IpRepr::Unspecified {
  377. src_addr: self.local_endpoint.addr,
  378. dst_addr: self.remote_endpoint.addr,
  379. protocol: IpProtocol::Tcp,
  380. };
  381. let mut repr = TcpRepr {
  382. src_port: self.local_endpoint.port,
  383. dst_port: self.remote_endpoint.port,
  384. control: TcpControl::None,
  385. seq_number: self.local_seq_no,
  386. ack_number: None,
  387. window_len: self.rx_buffer.window() as u16,
  388. payload: &[]
  389. };
  390. let ack_number = self.remote_seq_no + self.rx_buffer.len() as i32;
  391. match self.state {
  392. State::Closed | State::Listen => return Err(Error::Exhausted),
  393. State::SynReceived => {
  394. if !self.retransmit.check() { return Err(Error::Exhausted) }
  395. repr.control = TcpControl::Syn;
  396. net_trace!("tcp:{}:{}: sending SYN|ACK",
  397. self.local_endpoint, self.remote_endpoint);
  398. self.remote_last_ack = self.remote_seq_no;
  399. }
  400. State::Established => {
  401. if self.tx_buffer.len() > 0 && self.remote_win_len > 0 {
  402. if !self.retransmit.check() { return Err(Error::Exhausted) }
  403. // We can send something, so let's do that.
  404. let mut size = self.remote_win_len;
  405. // Clamp to MSS. Currently we only support the default MSS value.
  406. if size > 536 { size = 536 }
  407. // Extract data from the buffer. This may return less than what we want,
  408. // in case it's not possible to extract a contiguous slice.
  409. let data = self.tx_buffer.peek(size);
  410. net_trace!("tcp:{}:{}: sending {} octets",
  411. self.local_endpoint, self.remote_endpoint, data.len());
  412. repr.payload = data;
  413. } else if self.remote_last_ack != ack_number {
  414. // We don't have anything to send, or can't because the remote end does not
  415. // have any space to accept it, but we haven't yet acknowledged everything
  416. // we have received. So, do it.
  417. net_trace!("tcp:{}:{}: sending ACK",
  418. self.local_endpoint, self.remote_endpoint);
  419. } else {
  420. // We don't have anything to send and we've already acknowledged everything.
  421. return Err(Error::Exhausted)
  422. }
  423. }
  424. _ => unreachable!()
  425. }
  426. match self.state {
  427. // We don't have anything to acknowledge yet.
  428. State::Closed | State::Listen | State::SynSent => (),
  429. // Acknowledge all data we have received, since it is all in order.
  430. _ => {
  431. self.remote_last_ack = ack_number;
  432. repr.ack_number = Some(ack_number);
  433. }
  434. }
  435. emit(&ip_repr, &repr)
  436. }
  437. }
  438. impl<'a> IpPayload for TcpRepr<'a> {
  439. fn buffer_len(&self) -> usize {
  440. self.buffer_len()
  441. }
  442. fn emit(&self, ip_repr: &IpRepr, payload: &mut [u8]) {
  443. let mut packet = TcpPacket::new(payload).expect("undersized payload");
  444. self.emit(&mut packet, &ip_repr.src_addr(), &ip_repr.dst_addr())
  445. }
  446. }
  447. #[cfg(test)]
  448. mod test {
  449. use wire::IpAddress;
  450. use super::*;
  451. #[test]
  452. fn test_buffer() {
  453. let mut buffer = SocketBuffer::new(vec![0; 8]); // ........
  454. buffer.enqueue(6).copy_from_slice(b"foobar"); // foobar..
  455. assert_eq!(buffer.dequeue(3), b"foo"); // ...bar..
  456. buffer.enqueue(6).copy_from_slice(b"ba"); // ...barba
  457. buffer.enqueue(4).copy_from_slice(b"zho"); // zhobarba
  458. assert_eq!(buffer.dequeue(6), b"barba"); // zho.....
  459. assert_eq!(buffer.dequeue(8), b"zho"); // ........
  460. buffer.enqueue(8).copy_from_slice(b"gefug"); // ...gefug
  461. }
  462. #[test]
  463. fn test_buffer_wraparound() {
  464. let mut buffer = SocketBuffer::new(vec![0; 8]); // ........
  465. buffer.enqueue_slice(&b"foobar"[..]); // foobar..
  466. assert_eq!(buffer.dequeue(3), b"foo"); // ...bar..
  467. buffer.enqueue_slice(&b"bazhoge"[..]); // zhobarba
  468. }
  469. const LOCAL_IP: IpAddress = IpAddress::v4(10, 0, 0, 1);
  470. const REMOTE_IP: IpAddress = IpAddress::v4(10, 0, 0, 2);
  471. const LOCAL_PORT: u16 = 80;
  472. const REMOTE_PORT: u16 = 49500;
  473. const LOCAL_END: IpEndpoint = IpEndpoint::new(LOCAL_IP, LOCAL_PORT);
  474. const REMOTE_END: IpEndpoint = IpEndpoint::new(REMOTE_IP, REMOTE_PORT);
  475. const LOCAL_SEQ: i32 = 10000;
  476. const REMOTE_SEQ: i32 = -10000;
  477. const SEND_TEMPL: TcpRepr<'static> = TcpRepr {
  478. src_port: REMOTE_PORT, dst_port: LOCAL_PORT,
  479. control: TcpControl::None,
  480. seq_number: 0, ack_number: Some(0),
  481. window_len: 256, payload: &[]
  482. };
  483. const RECV_TEMPL: TcpRepr<'static> = TcpRepr {
  484. src_port: LOCAL_PORT, dst_port: REMOTE_PORT,
  485. control: TcpControl::None,
  486. seq_number: 0, ack_number: Some(0),
  487. window_len: 128, payload: &[]
  488. };
  489. fn send(socket: &mut TcpSocket, repr: &TcpRepr) -> Result<(), Error> {
  490. let mut buffer = vec![0; repr.buffer_len()];
  491. let mut packet = TcpPacket::new(&mut buffer).unwrap();
  492. repr.emit(&mut packet, &REMOTE_IP, &LOCAL_IP);
  493. let ip_repr = IpRepr::Unspecified {
  494. src_addr: REMOTE_IP,
  495. dst_addr: LOCAL_IP,
  496. protocol: IpProtocol::Tcp
  497. };
  498. socket.collect(&ip_repr, &packet.into_inner()[..])
  499. }
  500. fn recv<F>(socket: &mut TcpSocket, mut f: F)
  501. where F: FnMut(Result<TcpRepr, Error>) {
  502. let mut buffer = vec![];
  503. let result = socket.dispatch(&mut |ip_repr, payload| {
  504. assert_eq!(ip_repr.protocol(), IpProtocol::Tcp);
  505. assert_eq!(ip_repr.src_addr(), LOCAL_IP);
  506. assert_eq!(ip_repr.dst_addr(), REMOTE_IP);
  507. buffer.resize(payload.buffer_len(), 0);
  508. payload.emit(&ip_repr, &mut buffer[..]);
  509. let packet = TcpPacket::new(&buffer[..]).unwrap();
  510. let repr = try!(TcpRepr::parse(&packet, &ip_repr.src_addr(), &ip_repr.dst_addr()));
  511. Ok(f(Ok(repr)))
  512. });
  513. // Appease borrow checker.
  514. match result {
  515. Ok(()) => (),
  516. Err(e) => f(Err(e))
  517. }
  518. }
  519. macro_rules! send {
  520. ($socket:ident, [$( $repr:expr )*]) => ({
  521. $( send!($socket, $repr, Ok(())); )*
  522. });
  523. ($socket:ident, $repr:expr, $result:expr) =>
  524. (assert_eq!(send(&mut $socket, &$repr), $result))
  525. }
  526. macro_rules! recv {
  527. ($socket:ident, [$( $repr:expr )*]) => ({
  528. $( recv!($socket, Ok($repr)); )*
  529. recv!($socket, Err(Error::Exhausted))
  530. });
  531. ($socket:ident, $result:expr) =>
  532. (recv(&mut $socket, |repr| assert_eq!(repr, $result)))
  533. }
  534. fn init_logger() {
  535. extern crate log;
  536. use std::boxed::Box;
  537. struct Logger(());
  538. impl log::Log for Logger {
  539. fn enabled(&self, _metadata: &log::LogMetadata) -> bool {
  540. true
  541. }
  542. fn log(&self, record: &log::LogRecord) {
  543. println!("{}", record.args());
  544. }
  545. }
  546. let _ = log::set_logger(|max_level| {
  547. max_level.set(log::LogLevelFilter::Trace);
  548. Box::new(Logger(()))
  549. });
  550. println!("");
  551. }
  552. fn socket() -> TcpSocket<'static> {
  553. init_logger();
  554. let rx_buffer = SocketBuffer::new(vec![0; 128]);
  555. let tx_buffer = SocketBuffer::new(vec![0; 128]);
  556. match TcpSocket::new(rx_buffer, tx_buffer) {
  557. Socket::Tcp(socket) => socket,
  558. _ => unreachable!()
  559. }
  560. }
  561. // =========================================================================================//
  562. // Tests for the CLOSED state.
  563. // =========================================================================================//
  564. #[test]
  565. fn test_closed() {
  566. let mut s = socket();
  567. assert_eq!(s.state(), State::Closed);
  568. send!(s, TcpRepr {
  569. control: TcpControl::Syn,
  570. ..SEND_TEMPL
  571. }, Err(Error::Rejected));
  572. }
  573. // =========================================================================================//
  574. // Tests for the LISTEN state.
  575. // =========================================================================================//
  576. fn socket_listen() -> TcpSocket<'static> {
  577. let mut s = socket();
  578. s.state = State::Listen;
  579. s.local_endpoint = IpEndpoint::new(IpAddress::default(), LOCAL_PORT);
  580. s
  581. }
  582. #[test]
  583. fn test_listen_syn_no_ack() {
  584. let mut s = socket_listen();
  585. send!(s, TcpRepr {
  586. control: TcpControl::Syn,
  587. seq_number: REMOTE_SEQ,
  588. ack_number: Some(LOCAL_SEQ),
  589. ..SEND_TEMPL
  590. }, Err(Error::Malformed));
  591. assert_eq!(s.state, State::Listen);
  592. }
  593. #[test]
  594. fn test_listen_rst() {
  595. let mut s = socket_listen();
  596. send!(s, [TcpRepr {
  597. control: TcpControl::Rst,
  598. seq_number: REMOTE_SEQ,
  599. ack_number: None,
  600. ..SEND_TEMPL
  601. }]);
  602. }
  603. // =========================================================================================//
  604. // Tests for the SYN_RECEIVED state.
  605. // =========================================================================================//
  606. fn socket_syn_received() -> TcpSocket<'static> {
  607. let mut s = socket();
  608. s.state = State::SynReceived;
  609. s.local_endpoint = LOCAL_END;
  610. s.remote_endpoint = REMOTE_END;
  611. s.local_seq_no = LOCAL_SEQ;
  612. s.remote_seq_no = REMOTE_SEQ;
  613. s
  614. }
  615. #[test]
  616. fn test_syn_received_rst() {
  617. let mut s = socket_syn_received();
  618. send!(s, [TcpRepr {
  619. control: TcpControl::Rst,
  620. seq_number: REMOTE_SEQ,
  621. ack_number: Some(LOCAL_SEQ),
  622. ..SEND_TEMPL
  623. }]);
  624. assert_eq!(s.state, State::Listen);
  625. assert_eq!(s.local_endpoint, IpEndpoint::new(IpAddress::Unspecified, LOCAL_END.port));
  626. assert_eq!(s.remote_endpoint, IpEndpoint::default());
  627. }
  628. // =========================================================================================//
  629. // Tests for the SYN_SENT state.
  630. // =========================================================================================//
  631. fn socket_syn_sent() -> TcpSocket<'static> {
  632. let mut s = socket();
  633. s.state = State::SynSent;
  634. s.local_endpoint = LOCAL_END;
  635. s.remote_endpoint = REMOTE_END;
  636. s.local_seq_no = LOCAL_SEQ;
  637. s
  638. }
  639. #[test]
  640. fn test_syn_sent_rst() {
  641. let mut s = socket_syn_sent();
  642. send!(s, [TcpRepr {
  643. control: TcpControl::Rst,
  644. seq_number: REMOTE_SEQ,
  645. ack_number: Some(LOCAL_SEQ),
  646. ..SEND_TEMPL
  647. }]);
  648. assert_eq!(s.state, State::Closed);
  649. }
  650. #[test]
  651. fn test_syn_sent_rst_no_ack() {
  652. let mut s = socket_syn_sent();
  653. send!(s, TcpRepr {
  654. control: TcpControl::Rst,
  655. seq_number: REMOTE_SEQ,
  656. ack_number: None,
  657. ..SEND_TEMPL
  658. }, Err(Error::Malformed));
  659. assert_eq!(s.state, State::SynSent);
  660. }
  661. #[test]
  662. fn test_syn_sent_rst_bad_ack() {
  663. let mut s = socket_syn_sent();
  664. send!(s, TcpRepr {
  665. control: TcpControl::Rst,
  666. seq_number: REMOTE_SEQ,
  667. ack_number: Some(1234),
  668. ..SEND_TEMPL
  669. }, Err(Error::Malformed));
  670. assert_eq!(s.state, State::SynSent);
  671. }
  672. // =========================================================================================//
  673. // Tests for the ESTABLISHED state.
  674. // =========================================================================================//
  675. fn socket_established() -> TcpSocket<'static> {
  676. let mut s = socket();
  677. s.state = State::Established;
  678. s.local_endpoint = LOCAL_END;
  679. s.remote_endpoint = REMOTE_END;
  680. s.local_seq_no = LOCAL_SEQ + 1;
  681. s.remote_seq_no = REMOTE_SEQ + 1;
  682. s.remote_win_len = 128;
  683. s
  684. }
  685. #[test]
  686. fn test_established_receive() {
  687. let mut s = socket_established();
  688. send!(s, [TcpRepr {
  689. seq_number: REMOTE_SEQ + 1,
  690. ack_number: Some(LOCAL_SEQ + 1),
  691. payload: &b"abcdef"[..],
  692. ..SEND_TEMPL
  693. }]);
  694. recv!(s, [TcpRepr {
  695. seq_number: LOCAL_SEQ + 1,
  696. ack_number: Some(REMOTE_SEQ + 1 + 6),
  697. window_len: 122,
  698. ..RECV_TEMPL
  699. }]);
  700. assert_eq!(s.rx_buffer.dequeue(6), &b"abcdef"[..]);
  701. }
  702. #[test]
  703. fn test_established_send() {
  704. let mut s = socket_established();
  705. s.tx_buffer.enqueue_slice(b"abcdef");
  706. recv!(s, [TcpRepr {
  707. seq_number: LOCAL_SEQ + 1,
  708. ack_number: Some(REMOTE_SEQ + 1),
  709. payload: &b"abcdef"[..],
  710. ..RECV_TEMPL
  711. }]);
  712. assert_eq!(s.tx_buffer.len(), 6);
  713. send!(s, [TcpRepr {
  714. seq_number: REMOTE_SEQ + 1,
  715. ack_number: Some(LOCAL_SEQ + 1 + 6),
  716. ..SEND_TEMPL
  717. }]);
  718. assert_eq!(s.tx_buffer.len(), 0);
  719. }
  720. #[test]
  721. fn test_established_no_ack() {
  722. let mut s = socket_established();
  723. send!(s, TcpRepr {
  724. seq_number: REMOTE_SEQ + 1,
  725. ack_number: None,
  726. ..SEND_TEMPL
  727. }, Err(Error::Malformed));
  728. }
  729. #[test]
  730. fn test_established_bad_ack() {
  731. let mut s = socket_established();
  732. // Already acknowledged data.
  733. send!(s, TcpRepr {
  734. seq_number: REMOTE_SEQ + 1,
  735. ack_number: Some(LOCAL_SEQ - 1),
  736. ..SEND_TEMPL
  737. }, Err(Error::Malformed));
  738. assert_eq!(s.local_seq_no, LOCAL_SEQ + 1);
  739. // Data not yet transmitted.
  740. send!(s, TcpRepr {
  741. seq_number: REMOTE_SEQ + 1,
  742. ack_number: Some(LOCAL_SEQ + 10),
  743. ..SEND_TEMPL
  744. }, Err(Error::Malformed));
  745. assert_eq!(s.local_seq_no, LOCAL_SEQ + 1);
  746. }
  747. #[test]
  748. fn test_established_bad_seq() {
  749. let mut s = socket_established();
  750. // Data outside of receive window.
  751. send!(s, TcpRepr {
  752. seq_number: REMOTE_SEQ + 1 + 256,
  753. ack_number: Some(LOCAL_SEQ + 1),
  754. ..SEND_TEMPL
  755. }, Err(Error::Malformed));
  756. assert_eq!(s.remote_seq_no, REMOTE_SEQ + 1);
  757. }
  758. #[test]
  759. fn test_established_rst() {
  760. let mut s = socket_established();
  761. send!(s, [TcpRepr {
  762. control: TcpControl::Rst,
  763. seq_number: REMOTE_SEQ + 1,
  764. ack_number: Some(LOCAL_SEQ + 1),
  765. ..SEND_TEMPL
  766. }]);
  767. assert_eq!(s.state, State::Closed);
  768. }
  769. // =========================================================================================//
  770. // Tests for transitioning through multiple states.
  771. // =========================================================================================//
  772. #[test]
  773. fn test_listen() {
  774. let mut s = socket();
  775. s.listen(IpEndpoint::new(IpAddress::default(), LOCAL_PORT));
  776. assert_eq!(s.state, State::Listen);
  777. }
  778. #[test]
  779. fn test_three_way_handshake() {
  780. let mut s = socket();
  781. s.state = State::Listen;
  782. s.local_endpoint = IpEndpoint::new(IpAddress::default(), LOCAL_PORT);
  783. send!(s, [TcpRepr {
  784. control: TcpControl::Syn,
  785. seq_number: REMOTE_SEQ,
  786. ack_number: None,
  787. ..SEND_TEMPL
  788. }]);
  789. assert_eq!(s.state(), State::SynReceived);
  790. assert_eq!(s.local_endpoint(), LOCAL_END);
  791. assert_eq!(s.remote_endpoint(), REMOTE_END);
  792. recv!(s, [TcpRepr {
  793. control: TcpControl::Syn,
  794. seq_number: LOCAL_SEQ,
  795. ack_number: Some(REMOTE_SEQ + 1),
  796. ..RECV_TEMPL
  797. }]);
  798. send!(s, [TcpRepr {
  799. seq_number: REMOTE_SEQ + 1,
  800. ack_number: Some(LOCAL_SEQ + 1),
  801. ..SEND_TEMPL
  802. }]);
  803. assert_eq!(s.state(), State::Established);
  804. assert_eq!(s.local_seq_no, LOCAL_SEQ + 1);
  805. assert_eq!(s.remote_seq_no, REMOTE_SEQ + 1);
  806. }
  807. }