ihe-atna 0.4.0

// SPDX-FileCopyrightText: Cadasto B.V.
// SPDX-License-Identifier: BUSL-1.1

//! The connection to an Audit Record Repository: syslog over TLS (RFC 5425),
//! the transport ITI TF-2 §3.20.4.1.2.1.1 recommends.
//!
//! A repository is addressed as `tls://host[:port]`, the port defaulting to
//! 6514. Plain TCP (`tcp://host:port`, RFC 6587 octet counting) exists only
//! through [`Repository::unencrypted_for_development`]: an audit message
//! names the patient, so it never crosses a network in clear text outside a
//! development deployment.
//!
//! # Examples
//!
//! The syslog sender: a spool that stores every message first, the
//! repository it delivers to, and the forwarder that drains one into the
//! other.
//!
//! ```no_run
//! use std::path::Path;
//! use std::sync::Arc;
//! use std::time::Duration;
//!
//! use ihe_atna::atna::codes;
//! use ihe_atna::atna::forwarder::Forwarder;
//! use ihe_atna::atna::message::{
//!     ActiveParticipant, AuditMessage, AuditSource, CodedValue, EventIdentification,
//! };
//! use ihe_atna::atna::repository::{Repository, Timeouts, TlsSettings};
//! use ihe_atna::atna::spool::{Bounds, Spool};
//! use ihe_atna::atna::syslog::Sender;
//! use url::Url;
//!
//! # #[tokio::main(flavor = "current_thread")]
//! # async fn main() -> Result<(), Box<dyn std::error::Error>> {
//! let timeouts = Timeouts {
//!     connect: Duration::from_secs(5),
//!     send: Duration::from_secs(5),
//! };
//! let repository = Repository::tls(
//!     &Url::parse("tls://arr.example.org")?,
//!     &TlsSettings::default(),
//!     timeouts,
//! )?;
//! let bounds = Bounds {
//!     max_messages: 10_000,
//!     max_bytes: 64 << 20,
//!     write_timeout: Duration::from_secs(2),
//! };
//! let spool = Spool::open(Path::new("/var/spool/audit"), bounds)?;
//! let forwarder = Forwarder::new(spool, repository, Duration::from_secs(60));
//! tokio::spawn(Arc::clone(&forwarder).run());
//!
//! let message = AuditMessage {
//!     event: EventIdentification {
//!         event_id: CodedValue::of(codes::EVENT_APPLICATION_ACTIVITY),
//!         action: "E",
//!         date_time: jiff::Timestamp::now(),
//!         outcome: "0",
//!         type_codes: vec![CodedValue::of(codes::TYPE_APPLICATION_START)],
//!         outcome_description: None,
//!         purposes_of_use: Vec::new(),
//!     },
//!     participants: vec![ActiveParticipant {
//!         user_id: "gateway".to_owned(),
//!         alternative_user_id: None,
//!         user_name: None,
//!         user_is_requestor: false,
//!         role_id_codes: vec![CodedValue::of(codes::ROLE_APPLICATION)],
//!         access_point: None,
//!     }],
//!     source: AuditSource {
//!         id: "gateway.example.org".to_owned(),
//!         enterprise_site: None,
//!         type_codes: Vec::new(),
//!     },
//!     objects: Vec::new(),
//! };
//! let sender = Sender::new("gateway.example.org", "gateway", "4242")?;
//! let frame = sender.frame(jiff::Timestamp::now(), &message.to_xml()?);
//! forwarder.submit(frame).await?;
//! # Ok(())
//! # }
//! ```

use std::fmt;
use std::pin::Pin;
use std::sync::Arc;
use std::task::{Context, Poll, Waker};
use std::time::Duration;

use rustls::ClientConfig;
use rustls::pki_types::pem::PemObject as _;
use rustls::pki_types::{CertificateDer, PrivateKeyDer, ServerName};
use rustls_platform_verifier::Verifier;
use secrecy::{ExposeSecret, SecretSlice, SecretString};
use tokio::io::{AsyncRead, AsyncReadExt as _, AsyncWrite, AsyncWriteExt as _, ReadBuf};
use tokio::net::TcpStream;
use tokio_rustls::TlsConnector;
use url::Url;

use super::syslog::TLS_PORT;

/// Why a repository address or its TLS settings were refused.
#[derive(Debug, thiserror::Error)]
#[non_exhaustive]
pub enum RepositoryError {
    /// The address is not `tls://host[:port]`, or `tcp://host:port` for
    /// development.
    #[error("the audit repository address is not tls://host[:port]")]
    Address,
    /// The address has a path, a query, a fragment or userinfo.
    #[error("the audit repository address carries more than a host and a port")]
    Extra,
    /// The host is no TLS server name.
    #[error("the audit repository host is not a TLS server name")]
    ServerName(#[source] rustls::pki_types::InvalidDnsNameError),
    /// The trust roots do not read as PEM certificates.
    #[error("the audit repository trust roots are not PEM certificates")]
    Roots(#[source] rustls::pki_types::pem::Error),
    /// The client certificate and key do not read as PEM.
    #[error("the audit repository client certificate and key are not PEM")]
    Identity(#[source] rustls::pki_types::pem::Error),
    /// The client identity holds a key and no certificate.
    #[error("the audit repository client identity holds no certificate")]
    NoCertificate,
    /// rustls refused the configuration.
    #[error("the TLS configuration for the audit repository could not be built")]
    Tls(#[source] rustls::Error),
}

/// Why a frame was not written.
#[derive(Debug, thiserror::Error)]
#[non_exhaustive]
pub enum SendError {
    /// The connection could not be opened.
    #[error("the audit repository could not be reached")]
    Connect(#[source] std::io::Error),
    /// The TLS handshake failed.
    #[error("the TLS handshake with the audit repository failed")]
    Handshake(#[source] std::io::Error),
    /// A step did not finish within its timeout.
    #[error("the audit repository did not finish the {0} within its timeout")]
    Timeout(Step),
    /// The frame could not be written.
    #[error("the audit message could not be written to the audit repository")]
    Write(#[source] std::io::Error),
}

/// A step of sending a frame, each bounded by a timeout.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[non_exhaustive]
pub enum Step {
    /// Opening the TCP connection.
    Connect,
    /// The TLS handshake.
    Handshake,
    /// Writing the frame.
    Write,
    /// Flushing it.
    Flush,
    /// Closing the connection: the TLS `close_notify` and the repository's
    /// end of the stream.
    Close,
}

impl fmt::Display for Step {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.write_str(match self {
            Self::Connect => "connection",
            Self::Handshake => "TLS handshake",
            Self::Write => "write",
            Self::Flush => "flush",
            Self::Close => "close",
        })
    }
}

/// How long each step of sending a frame may take.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct Timeouts {
    /// Opening the TCP connection.
    pub connect: Duration,
    /// The TLS handshake, and each write and flush of a frame.
    pub send: Duration,
}

/// How a connection is secured.
#[derive(Clone)]
enum Transport {
    Tls(TlsConnector, ServerName<'static>),
    Plain,
}

/// An Audit Record Repository the sender writes to.
#[derive(Clone)]
pub struct Repository {
    host: String,
    port: u16,
    transport: Transport,
    timeouts: Timeouts,
}

impl fmt::Debug for Repository {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("Repository")
            .field("host", &self.host)
            .field("port", &self.port)
            .field("tls", &matches!(self.transport, Transport::Tls(..)))
            .finish_non_exhaustive()
    }
}

/// The TLS settings of a repository connection.
#[derive(Debug, Clone, Default)]
pub struct TlsSettings {
    /// PEM certificates trusted beside the platform's roots.
    pub roots: Option<Vec<u8>>,
    /// The PEM client certificate chain and private key the sender
    /// authenticates with, when the repository asks for one.
    pub identity: Option<SecretString>,
}

impl Repository {
    /// The repository at `address`, `tls://host[:port]`, reached over TLS
    /// 1.2 or later with `tls`, each step bounded by `timeouts`.
    ///
    /// # Errors
    ///
    /// A [`RepositoryError`] for an address of another form, and for trust
    /// roots or an identity that do not read as PEM.
    pub fn tls(
        address: &Url,
        tls: &TlsSettings,
        timeouts: Timeouts,
    ) -> Result<Self, RepositoryError> {
        if address.scheme() != "tls" {
            return Err(RepositoryError::Address);
        }
        let (host, port) = host_port(address, Some(TLS_PORT))?;
        let name = ServerName::try_from(host.clone()).map_err(RepositoryError::ServerName)?;
        let connector = TlsConnector::from(Arc::new(client_config(tls)?));
        Ok(Self {
            host,
            port,
            transport: Transport::Tls(connector, name),
            timeouts,
        })
    }

    /// The repository at `address`, `tcp://host:port`, reached without
    /// encryption: for a development deployment only, since every message
    /// names a patient.
    ///
    /// # Errors
    ///
    /// [`RepositoryError::Address`] for an address of another form.
    pub fn unencrypted_for_development(
        address: &Url,
        timeouts: Timeouts,
    ) -> Result<Self, RepositoryError> {
        if address.scheme() != "tcp" {
            return Err(RepositoryError::Address);
        }
        let (host, port) = host_port(address, None)?;
        Ok(Self {
            host,
            port,
            transport: Transport::Plain,
            timeouts,
        })
    }

    /// Opens a connection: the TCP connection within the connect timeout,
    /// then the TLS handshake within the send timeout.
    ///
    /// # Errors
    ///
    /// A [`SendError`] when the repository cannot be reached, fails the TLS
    /// handshake, or does not finish a step within its timeout.
    pub async fn connect(&self) -> Result<Connection, SendError> {
        let stream = bounded(
            self.timeouts.connect,
            Step::Connect,
            TcpStream::connect((self.host.as_str(), self.port)),
        )
        .await?
        .map_err(SendError::Connect)?;
        // NOTE: no specification governs this: our own design; a frame goes out at
        // once, so it never waits on the acknowledgement of the one before it.
        stream.set_nodelay(true).map_err(SendError::Connect)?;
        let stream: Box<dyn Stream> = match &self.transport {
            Transport::Plain => Box::new(stream),
            Transport::Tls(connector, name) => Box::new(
                bounded(
                    self.timeouts.send,
                    Step::Handshake,
                    connector.connect(name.clone(), stream),
                )
                .await?
                .map_err(SendError::Handshake)?,
            ),
        };
        Ok(Connection {
            stream,
            send_timeout: self.timeouts.send,
        })
    }
}

/// `step`, given at most `limit` to finish.
async fn bounded<T>(
    limit: Duration,
    step: Step,
    future: impl Future<Output = T>,
) -> Result<T, SendError> {
    tokio::time::timeout(limit, future)
        .await
        .map_err(|_elapsed| SendError::Timeout(step))
}

/// A connection's byte stream, plain or TLS; both are `Sync`, so a
/// connection held behind a shared reference needs no lock of its own.
trait Stream: AsyncRead + AsyncWrite + Send + Sync + Unpin {}

impl<T: AsyncRead + AsyncWrite + Send + Sync + Unpin> Stream for T {}

/// An open connection to a repository, `Send` and `Sync`.
pub struct Connection {
    stream: Box<dyn Stream>,
    send_timeout: Duration,
}

impl fmt::Debug for Connection {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("Connection").finish_non_exhaustive()
    }
}

impl Connection {
    /// Whether the repository has closed the connection, or it has failed.
    ///
    /// A repository sends nothing on a syslog connection (RFC 5425 §4.3), so
    /// a read that ends or fails at once means the connection is gone. A
    /// write to a connection closed after this check can still be lost:
    /// syslog has no acknowledgement.
    ///
    /// The read is polled once and never waited on, so a live connection
    /// answers at once rather than after a timer tick. The yield before it
    /// lets the runtime read the socket's readiness first, so an end the
    /// repository sent is seen.
    pub async fn is_closed(&mut self) -> bool {
        tokio::task::yield_now().await;
        let mut probe = [0_u8; 1];
        let mut buf = ReadBuf::new(&mut probe);
        let mut cx = Context::from_waker(Waker::noop());
        match Pin::new(&mut self.stream).poll_read(&mut cx, &mut buf) {
            Poll::Ready(Ok(())) => buf.filled().is_empty(),
            Poll::Ready(Err(_)) => true,
            Poll::Pending => false,
        }
    }

    /// Writes `frame` and flushes it, each within the send timeout, so a
    /// repository that stops reading cannot hold the sender.
    ///
    /// # Errors
    ///
    /// [`SendError::Write`] when the connection fails, and
    /// [`SendError::Timeout`] when the write or the flush does not finish in
    /// time; the connection is not used again after either.
    pub async fn send(&mut self, frame: &SecretSlice<u8>) -> Result<(), SendError> {
        let limit = self.send_timeout;
        bounded(
            limit,
            Step::Write,
            self.stream.write_all(frame.expose_secret()),
        )
        .await?
        .map_err(SendError::Write)?;
        bounded(limit, Step::Flush, self.stream.flush())
            .await?
            .map_err(SendError::Write)
    }

    /// Ends the connection as RFC 5425 §4.4 asks of a transport sender: a
    /// TLS `close_notify` with the write half shut, then the stream read to
    /// the repository's end, each within the send timeout.
    ///
    /// Dropping a connection instead closes the socket with the repository's
    /// TLS session tickets unread, and the kernel then resets it, which can
    /// discard frames [`Connection::send`] already handed over. Close a
    /// connection whenever no further frame will follow on it.
    ///
    /// # Errors
    ///
    /// [`SendError::Write`] when the close cannot be written or the
    /// connection fails while it ends, and [`SendError::Timeout`] with
    /// [`Step::Close`] when the repository does not end the stream in time;
    /// a frame sent last may then be lost.
    pub async fn close(mut self) -> Result<(), SendError> {
        let limit = self.send_timeout;
        bounded(limit, Step::Close, self.stream.shutdown())
            .await?
            .map_err(SendError::Write)?;
        let stream = &mut self.stream;
        bounded(limit, Step::Close, async move {
            let mut unread = [0_u8; 256];
            loop {
                match stream.read(&mut unread).await {
                    Ok(0) => return Ok(()),
                    Ok(_) => {}
                    // NOTE: RFC 5425 §4.4 lets a receiver close without its own
                    // close_notify, an incomplete close that still ends the stream.
                    Err(error) if error.kind() == std::io::ErrorKind::UnexpectedEof => {
                        return Ok(());
                    }
                    Err(error) => return Err(error),
                }
            }
        })
        .await?
        .map_err(SendError::Write)
    }
}

/// The host and port of `address`, the port `default` when it names none.
fn host_port(address: &Url, default: Option<u16>) -> Result<(String, u16), RepositoryError> {
    let clean = address.path().is_empty()
        && address.query().is_none()
        && address.fragment().is_none()
        && address.username().is_empty()
        && address.password().is_none();
    if !clean {
        return Err(RepositoryError::Extra);
    }
    let host = match address.host() {
        Some(url::Host::Domain(name)) => name.to_owned(),
        Some(url::Host::Ipv4(ip)) => ip.to_string(),
        Some(url::Host::Ipv6(ip)) => ip.to_string(),
        None => return Err(RepositoryError::Address),
    };
    let port = address.port().or(default).ok_or(RepositoryError::Address)?;
    Ok((host, port))
}

/// The client configuration: rustls over aws-lc-rs at TLS 1.2 or later,
/// verifying the repository against the platform's roots and `tls.roots`.
fn client_config(tls: &TlsSettings) -> Result<ClientConfig, RepositoryError> {
    let provider = Arc::new(rustls::crypto::aws_lc_rs::default_provider());
    let roots = match &tls.roots {
        Some(pem) => CertificateDer::pem_slice_iter(pem)
            .collect::<Result<Vec<_>, _>>()
            .map_err(RepositoryError::Roots)?,
        None => Vec::new(),
    };
    let verifier = Verifier::new_with_extra_roots(roots, Arc::clone(&provider))
        .map_err(RepositoryError::Tls)?;
    // NOTE: ITI TF-2 §3.20.4.1.2.1.1: the transport must be TLS, 1.2 recommended.
    let builder = ClientConfig::builder_with_provider(provider)
        .with_safe_default_protocol_versions()
        .map_err(RepositoryError::Tls)?
        .dangerous()
        .with_custom_certificate_verifier(Arc::new(verifier));
    match &tls.identity {
        None => Ok(builder.with_no_client_auth()),
        Some(pem) => {
            let pem = pem.expose_secret().as_bytes();
            let chain = CertificateDer::pem_slice_iter(pem)
                .collect::<Result<Vec<_>, _>>()
                .map_err(RepositoryError::Identity)?;
            let key = PrivateKeyDer::from_pem_slice(pem).map_err(RepositoryError::Identity)?;
            if chain.is_empty() {
                return Err(RepositoryError::NoCertificate);
            }
            builder
                .with_client_auth_cert(chain, key)
                .map_err(RepositoryError::Tls)
        }
    }
}

#[cfg(test)]
mod tests {
    use super::Connection;

    const fn is_send_and_sync<T: Send + Sync>() {}

    #[test]
    fn a_connection_is_send_and_sync() {
        const { is_send_and_sync::<Connection>() };
    }
}