UDP Tracker client: Print unrecognized responses

[josecelano]

Parent issue: #669

When you run the UDP tracker client:

cargo run --bin udp_tracker_client announce 127.0.0.1:6969 9c38422213e30bff212b30c360d26f9a02136422

You could receive a response that can't be parsed into an aquatic UDP response. In this case, we should print the response.

This implies changing the UdtpTrackerClient to return an error with the received packet.

#[allow(clippy::module_name_repetitions)]
#[derive(Debug)]
pub struct UdpTrackerClient {
    pub udp_client: UdpClient,
}

impl UdpTrackerClient {
    /// # Panics
    ///
    /// Will panic if can't write request to bytes.
    pub async fn send(&self, request: Request) -> usize {
        debug!(target: "UDP tracker client", "send request {request:?}");

        // Write request into a buffer
        let request_buffer = vec![0u8; MAX_PACKET_SIZE];
        let mut cursor = Cursor::new(request_buffer);

        let request_data = match request.write(&mut cursor) {
            Ok(()) => {
                #[allow(clippy::cast_possible_truncation)]
                let position = cursor.position() as usize;
                let inner_request_buffer = cursor.get_ref();
                // Return slice which contains written request data
                &inner_request_buffer[..position]
            }
            Err(e) => panic!("could not write request to bytes: {e}."),
        };

        self.udp_client.send(request_data).await
    }

    /// # Panics
    ///
    /// Will panic if can't create response from the received payload (bytes buffer).
    pub async fn receive(&self) -> Response {
        let mut response_buffer = [0u8; MAX_PACKET_SIZE];

        let payload_size = self.udp_client.receive(&mut response_buffer).await;

        debug!(target: "UDP tracker client", "received {payload_size} bytes. Response {response_buffer:?}");

        Response::from_bytes(&response_buffer[..payload_size], true).unwrap()
    }
}

As you can see on this line:

Response::from_bytes(&response_buffer[..payload_size], true).unwrap()

The client would panic if it received a response that could be deserialized into a valid aquatic response.

We have to change the signature of the receive function to return a Result<Response, Error>. The error can contain the data received.

You can get a list of UDP trackers from https://newtrackon.com/. The client should print all the responses from all those trackers.

[josecelano]

Relates to: #814 (comment)

What I wanted to do is to include the bytes in the response in the error message. For example, this is a normal response:

cargo run --bin udp_tracker_client announce 127.0.0.1:6969 9c38422213e30bff212b30c360d26f9a02136422 | jq
    Finished `dev` profile [optimized + debuginfo] target(s) in 0.09s
     Running `target/debug/udp_tracker_client announce '127.0.0.1:6969' 9c38422213e30bff212b30c360d26f9a02136422`
{
  "transaction_id": -888840697,
  "announce_interval": 120,
  "leechers": 0,
  "seeders": 1,
  "peers": []
}

If you force the UDP tracker to return a stream of bytes that the aquatic package can't recognize (it can't convert it into an UDP response) then you have this error:

cargo run --bin udp_tracker_client announce 127.0.0.1:6969 9c38422213e30bff212b30c360d26f9a02136422
   Compiling torrust-tracker v3.0.0-alpha.12-develop (/home/josecelano/Documents/git/committer/me/github/torrust/torrust-tracker)
    Finished `dev` profile [optimized + debuginfo] target(s) in 2.57s
     Running `target/debug/udp_tracker_client announce '127.0.0.1:6969' 9c38422213e30bff212b30c360d26f9a02136422`
Error: failed to fill whole buffer

In that case, I have not provided enough bytes for an announce response.

This is how I've forced an invalid UDP announce response in the UDP handler:

    async fn send_response(socket: &Arc<UdpSocket>, to: SocketAddr, response: Response) {
        trace!("Sending Response: {response:?} to: {to:?}");

        let buffer = vec![0u8; MAX_PACKET_SIZE];
        let mut cursor = Cursor::new(buffer);

        match response.write(&mut cursor) {
            Ok(()) => {
                #[allow(clippy::cast_possible_truncation)]
                let position = cursor.position() as usize;
                let inner = cursor.get_ref();

                debug!("Sending {} bytes ...", &inner[..position].len());
                debug!("To: {:?}", &to);
                debug!("Payload: {:?}", &inner[..position]);

                match response {
                    Response::Connect(_) => Self::send_packet(socket, &to, &inner[..position]).await,
                    Response::AnnounceIpv4(_) => {
                        let changed_response = [0x00_00_00_01];
                        Self::send_packet(socket, &to, &changed_response).await
                    }
                    Response::AnnounceIpv6(_) => Self::send_packet(socket, &to, &inner[..position]).await,
                    Response::Scrape(_) => Self::send_packet(socket, &to, &inner[..position]).await,
                    Response::Error(_) => Self::send_packet(socket, &to, &inner[..position]).await,
                }

                debug!("{} bytes sent", &inner[..position].len());
            }
            Err(_) => {
                error!("could not write response to bytes.");
            }
        }
    }

However, I've just realized it's not possible to capture these errors because the aquatic package panics:

impl Response {
    #[inline]
    pub fn write(&self, bytes: &mut impl Write) -> Result<(), io::Error> {
        // ... 
    }
   
    #[inline]
    pub fn from_bytes(bytes: &[u8], ipv4: bool) -> Result<Self, io::Error> {
        let mut cursor = Cursor::new(bytes);

        let action = cursor.read_i32::<NetworkEndian>()?;
        let transaction_id = cursor.read_i32::<NetworkEndian>()?;

        match action {
            // Connect
            0 => {
                let connection_id = cursor.read_i64::<NetworkEndian>()?;

                Ok((ConnectResponse {
                    connection_id: ConnectionId(connection_id),
                    transaction_id: TransactionId(transaction_id),
                })
                .into())
            }
            // Announce
            1 if ipv4 => {
                let announce_interval = cursor.read_i32::<NetworkEndian>()?;
                let leechers = cursor.read_i32::<NetworkEndian>()?;
                let seeders = cursor.read_i32::<NetworkEndian>()?;

                let position = cursor.position() as usize;
                let inner = cursor.into_inner();

                let peers = inner[position..]
                    .chunks_exact(6)
                    .map(|chunk| {
                        let ip_bytes: [u8; 4] = (&chunk[..4]).try_into().unwrap();
                        let ip_address = Ipv4Addr::from(ip_bytes);
                        let port = (&chunk[4..]).read_u16::<NetworkEndian>().unwrap();

                        ResponsePeer {
                            ip_address,
                            port: Port(port),
                        }
                    })
                    .collect();

                Ok((AnnounceResponse {
                    transaction_id: TransactionId(transaction_id),
                    announce_interval: AnnounceInterval(announce_interval),
                    leechers: NumberOfPeers(leechers),
                    seeders: NumberOfPeers(seeders),
                    peers,
                })
                .into())
            }
            1 if !ipv4 => {
                let announce_interval = cursor.read_i32::<NetworkEndian>()?;
                let leechers = cursor.read_i32::<NetworkEndian>()?;
                let seeders = cursor.read_i32::<NetworkEndian>()?;

                let position = cursor.position() as usize;
                let inner = cursor.into_inner();

                let peers = inner[position..]
                    .chunks_exact(18)
                    .map(|chunk| {
                        let ip_bytes: [u8; 16] = (&chunk[..16]).try_into().unwrap();
                        let ip_address = Ipv6Addr::from(ip_bytes);
                        let port = (&chunk[16..]).read_u16::<NetworkEndian>().unwrap();

                        ResponsePeer {
                            ip_address,
                            port: Port(port),
                        }
                    })
                    .collect();

                Ok((AnnounceResponse {
                    transaction_id: TransactionId(transaction_id),
                    announce_interval: AnnounceInterval(announce_interval),
                    leechers: NumberOfPeers(leechers),
                    seeders: NumberOfPeers(seeders),
                    peers,
                })
                .into())
            }
            // Scrape
            2 => {
                let position = cursor.position() as usize;
                let inner = cursor.into_inner();

                let stats = inner[position..]
                    .chunks_exact(12)
                    .map(|chunk| {
                        let mut cursor: Cursor<&[u8]> = Cursor::new(&chunk[..]);

                        let seeders = cursor.read_i32::<NetworkEndian>().unwrap();
                        let downloads = cursor.read_i32::<NetworkEndian>().unwrap();
                        let leechers = cursor.read_i32::<NetworkEndian>().unwrap();

                        TorrentScrapeStatistics {
                            seeders: NumberOfPeers(seeders),
                            completed: NumberOfDownloads(downloads),
                            leechers: NumberOfPeers(leechers),
                        }
                    })
                    .collect();

                Ok((ScrapeResponse {
                    transaction_id: TransactionId(transaction_id),
                    torrent_stats: stats,
                })
                .into())
            }
            // Error
            3 => {
                let position = cursor.position() as usize;
                let inner = cursor.into_inner();

                Ok((ErrorResponse {
                    transaction_id: TransactionId(transaction_id),
                    message: String::from_utf8_lossy(&inner[position..])
                        .into_owned()
                        .into(),
                })
                .into())
            }
            _ => Ok((ErrorResponse {
                transaction_id: TransactionId(transaction_id),
                message: "Invalid action".into(),
            })
            .into()),
        }
    }
}

In some cases, it's returning an error, but in other cases, it's unwrapping.

@greatest-ape I think it would be great to return an error in all cases. In fact, for the same case (constructing an i32 cursor.read_i32::<NetworkEndian>(), sometimes it returns an error and sometimes unwraps).

In the end, I just want to display the received package as a byte stream so the debugger can find out what happened.

@ngthhu, for the time being, we could do it only for errors we can capture. For example, the sixth field in the response is an IP:

https://docs.rs/torrust-tracker/3.0.0-alpha.11/torrust_tracker/servers/udp/index.html#announce-response

We can return something that's no a valid IP.

I would expect to see something like:

Error: Unrecognized UDP tracker response. Expected a valid UDP response, got: [123, 2323, 44545, 343, ...]

[greatest-ape]

When exactly does it panic?

Also, on a side note, I’m planning to upload a new aquatic release to crates.io very soon, if that is what is preventing you from using the latest protocol code :-)

[josecelano]

When exactly does it panic?

Also, on a side note, I’m planning to upload a new aquatic release to crates.io very soon, if that is what is preventing you from using the latest protocol code :-)

Hey @greatest-ape sorry, The modified handler to force the error was wrong:

Response::AnnounceIpv4(_) => {
    let changed_response = [0x00_00_00_01];
    Self::send_packet(socket, &to, &changed_response).await
}

I was sending just 4 bytes (one i32). So the error must be in the second field transaction_id:

let action = cursor.read_i32::<NetworkEndian>()?;
let transaction_id = cursor.read_i32::<NetworkEndian>()?;

There are no more bytes to read.

In this case, it's not an unwrap, and we can catch the error. It seems the problem is only parsing IPs and ports.

And good to know there will be a new version for aquatic. I'm looking forward to seeing what has changed.

[greatest-ape]

I don’t get it, does it still panic? If it does, where exactly? Parsing IPs and ports shouldn’t panic either, due to chunks_exact.

[greatest-ape]

Anyway, a new aquatic_udp_protocol is on crates.io now. I actually added tests of parsing requests from various length byte vectors and could not produce a panic.