libdd-crashtracker 3.0.0

// Copyright 2023-Present Datadog, Inc. https://www.datadoghq.com/
// SPDX-License-Identifier: Apache-2.0
#![cfg(unix)]
#![allow(clippy::std_instead_of_alloc, clippy::std_instead_of_core)]

mod debug_logger;
mod entry_points;
pub use entry_points::{
    async_receiver_entry_point_stream, async_receiver_entry_point_unix_listener,
    async_receiver_entry_point_unix_socket, get_receiver_unix_socket, receiver_entry_point_stdin,
    receiver_entry_point_unix_socket,
};
#[cfg(target_os = "linux")]
mod ptrace_collector;
mod receive_report;

#[cfg(feature = "benchmarking")]
pub mod benchmark;

#[cfg(test)]
mod tests {
    use super::receive_report::*;
    use crate::collector::default_signals;
    use crate::crash_info::{SiCodes, SigInfo, SignalNames};
    use crate::shared::constants::*;
    use crate::{CrashtrackerConfiguration, ErrorKind};
    use std::time::Duration;
    use tokio::io::{AsyncReadExt, AsyncWriteExt, BufReader};
    use tokio::net::UnixStream;

    async fn to_socket(
        target: &mut tokio::net::UnixStream,
        msg: impl AsRef<str>,
    ) -> anyhow::Result<usize> {
        let msg = msg.as_ref();
        let n = target.write(format!("{msg}\n").as_bytes()).await?;
        target.flush().await?;
        Ok(n)
    }

    async fn send_report(delay: Duration, mut stream: UnixStream) -> anyhow::Result<()> {
        let sender = &mut stream;
        send_report_prelude(sender).await?;
        tokio::time::sleep(delay).await;
        to_socket(sender, DD_CRASHTRACK_DONE).await?;
        Ok(())
    }

    async fn send_report_lines(sender: &mut UnixStream) -> anyhow::Result<()> {
        send_report_prelude(sender).await?;
        to_socket(sender, DD_CRASHTRACK_DONE).await?;
        Ok(())
    }

    async fn send_report_prelude(sender: &mut UnixStream) -> anyhow::Result<()> {
        to_socket(sender, DD_CRASHTRACK_BEGIN_SIGINFO).await?;
        to_socket(
            sender,
            serde_json::to_string(&SigInfo {
                si_addr: None,
                si_code: 2,
                si_code_human_readable: SiCodes::BUS_ADRALN,
                si_signo: 11,
                si_signo_human_readable: SignalNames::SIGSEGV,
            })?,
        )
        .await?;
        to_socket(sender, DD_CRASHTRACK_END_SIGINFO).await?;

        to_socket(sender, DD_CRASHTRACK_BEGIN_KIND).await?;
        to_socket(sender, serde_json::to_string(&ErrorKind::UnixSignal)?).await?;
        to_socket(sender, DD_CRASHTRACK_END_KIND).await?;

        to_socket(sender, DD_CRASHTRACK_BEGIN_CONFIG).await?;
        let builder = CrashtrackerConfiguration::builder();
        let config = builder
            .signals(default_signals())
            .timeout(Duration::from_secs(3))
            .use_alt_stack(true)
            .build()?;
        to_socket(sender, serde_json::to_string(&config)?).await?;
        to_socket(sender, DD_CRASHTRACK_END_CONFIG).await?;
        Ok(())
    }

    #[tokio::test]
    #[cfg_attr(miri, ignore)]
    async fn test_receive_report_short_timeout() -> anyhow::Result<()> {
        let (sender, receiver) = tokio::net::UnixStream::pair()?;

        let join_handle1 = tokio::spawn(async move {
            let mut stream = BufReader::new(receiver);
            receive_report_from_stream(Duration::from_secs(1), &mut stream).await
        });
        let join_handle2 = tokio::spawn(send_report(Duration::from_secs(2), sender));

        let crash_report = join_handle1.await??;
        let (_config, crashinfo) = crash_report.expect("Expect a report");
        assert!(crashinfo.incomplete);
        let sender_error = join_handle2.await?.unwrap_err().to_string();
        assert_eq!(sender_error, "Broken pipe (os error 32)");
        Ok(())
    }

    #[tokio::test]
    #[cfg_attr(miri, ignore)]
    async fn test_receive_report_long_timeout() -> anyhow::Result<()> {
        let (sender, receiver) = tokio::net::UnixStream::pair()?;

        let join_handle1 = tokio::spawn(async move {
            let mut stream = BufReader::new(receiver);
            receive_report_from_stream(Duration::from_secs(2), &mut stream).await
        });
        let join_handle2 = tokio::spawn(send_report(Duration::from_secs(1), sender));

        let crash_report = join_handle1.await??;
        let (_config, crashinfo) = crash_report.expect("Expect a report");
        assert!(crashinfo.incomplete);
        join_handle2.await??;
        Ok(())
    }

    /// The collector blocks on this connection until it sees a hangup, and the crashing
    /// process is torn down as soon as it unblocks. Normalization and symbolization read
    /// `/proc/<pid>/maps` and the process' mapped files, so returning the report must not
    /// close the connection: the caller has to keep it open until symbolization is done.
    #[tokio::test]
    #[cfg_attr(miri, ignore)]
    async fn test_receive_report_keeps_collector_connection_open() -> anyhow::Result<()> {
        let (mut sender, receiver) = tokio::net::UnixStream::pair()?;

        // The whole report fits in the socket buffer, so it can be written up front and
        // the receiver can be driven from this task while the sender end stays owned here.
        send_report_lines(&mut sender).await?;

        let mut stream = BufReader::new(receiver);
        let crash_report = receive_report_from_stream(Duration::from_secs(5), &mut stream).await?;
        assert!(crash_report.is_some(), "Expect a report");

        let mut buf = [0u8; 1];
        let hangup = tokio::time::timeout(Duration::from_millis(200), sender.read(&mut buf)).await;
        assert!(
            hangup.is_err(),
            "receive_report_from_stream released the crashing process before symbolization"
        );

        // Dropping the receiving end is what releases the collector.
        drop(stream);
        let n = tokio::time::timeout(Duration::from_secs(5), sender.read(&mut buf)).await??;
        assert_eq!(n, 0, "Expected a hangup once the receiver drops the stream");
        Ok(())
    }
}