use std::sync::Arc; use std::time::Duration; use anyhow::Result; use assert_matches::*; use prometheus::Registry as MetricsRegistry; use quick_xml::events::Event; use quick_xml::NsReader; use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; use tokio::io::{ReadHalf as GenericReadHalf, WriteHalf as GenericWriteHalf}; use tokio::net::tcp::{ReadHalf, WriteHalf}; use tokio::net::TcpStream; use tokio_rustls::client::TlsStream; use tokio_rustls::rustls::client::ServerCertVerifier; use tokio_rustls::rustls::{ClientConfig, ServerName}; use tokio_rustls::TlsConnector; use lavina_core::player::PlayerRegistry; use lavina_core::repo::{Storage, StorageConfig}; use lavina_core::room::RoomRegistry; use projection_xmpp::{launch, ServerConfig}; use proto_xmpp::xml::{Continuation, FromXml, Parser}; pub async fn read_irc_message(reader: &mut BufReader>, buf: &mut Vec) -> Result { let mut size = 0; let res = reader.read_until(b'\n', buf).await?; size += res; return Ok(size); } struct TestScope<'a> { reader: NsReader>>, writer: WriteHalf<'a>, buffer: Vec, } impl<'a> TestScope<'a> { fn new(stream: &mut TcpStream) -> TestScope<'_> { let (reader, writer) = stream.split(); let reader = NsReader::from_reader(BufReader::new(reader)); let buffer = vec![]; TestScope { reader, writer, buffer } } async fn send(&mut self, str: &str) -> Result<()> { self.writer.write_all(str.as_bytes()).await?; self.writer.write_all(b"\n").await?; self.writer.flush().await?; Ok(()) } async fn next_xml_event(&mut self) -> Result> { self.buffer.clear(); let event = self.reader.read_event_into_async(&mut self.buffer).await?; Ok(event) } async fn read(&mut self) -> Result { self.buffer.clear(); let (ns, event) = self.reader.read_resolved_event_into_async(&mut self.buffer).await?; let mut parser: Continuation<_, std::result::Result> = T::parse().consume(ns, &event); loop { match parser { Continuation::Final(res) => return Ok(res?), Continuation::Continue(next) => { let (ns, event) = self.reader.read_resolved_event_into_async(&mut self.buffer).await?; parser = next.consume(ns, &event); } } } } } struct TestScopeTls<'a> { reader: NsReader>>>, writer: GenericWriteHalf<&'a mut TlsStream>, buffer: Vec, pub timeout: Duration, } impl<'a> TestScopeTls<'a> { fn new(stream: &'a mut TlsStream, buffer: Vec) -> TestScopeTls<'a> { let (reader, writer) = tokio::io::split(stream); let reader = NsReader::from_reader(BufReader::new(reader)); let timeout = Duration::from_millis(100); TestScopeTls { reader, writer, buffer, timeout, } } async fn send(&mut self, str: &str) -> Result<()> { self.writer.write_all(str.as_bytes()).await?; self.writer.write_all(b"\n").await?; self.writer.flush().await?; Ok(()) } async fn next_xml_event(&mut self) -> Result> { self.buffer.clear(); let event = self.reader.read_event_into_async(&mut self.buffer); let event = tokio::time::timeout(self.timeout, event).await??; Ok(event) } } struct IgnoreCertVerification; impl ServerCertVerifier for IgnoreCertVerification { fn verify_server_cert( &self, _end_entity: &tokio_rustls::rustls::Certificate, _intermediates: &[tokio_rustls::rustls::Certificate], _server_name: &ServerName, _scts: &mut dyn Iterator, _ocsp_response: &[u8], _now: std::time::SystemTime, ) -> std::result::Result { Ok(tokio_rustls::rustls::client::ServerCertVerified::assertion()) } } #[tokio::test] async fn scenario_basic() -> Result<()> { tracing_subscriber::fmt::init(); let config = ServerConfig { listen_on: "127.0.0.1:0".parse().unwrap(), cert: "tests/certs/xmpp.pem".parse().unwrap(), key: "tests/certs/xmpp.key".parse().unwrap(), }; let mut metrics = MetricsRegistry::new(); let mut storage = Storage::open(StorageConfig { db_path: ":memory:".into(), }) .await?; let rooms = RoomRegistry::new(&mut metrics, storage.clone()).unwrap(); let players = PlayerRegistry::empty(rooms.clone(), &mut metrics).unwrap(); let server = launch(config, players, rooms, metrics, storage.clone()).await.unwrap(); // test scenario storage.create_user("tester").await?; storage.set_password("tester", "password").await?; let mut stream = TcpStream::connect(server.addr).await?; let mut s = TestScope::new(&mut stream); tracing::info!("TCP connection established"); s.send(r#""#).await?; s.send(r#""#).await?; assert_matches!(s.next_xml_event().await?, Event::Decl(_) => {}); assert_matches!(s.next_xml_event().await?, Event::Start(b) => assert_eq!(b.local_name().into_inner(), b"stream")); assert_matches!(s.next_xml_event().await?, Event::Start(b) => assert_eq!(b.local_name().into_inner(), b"features")); assert_matches!(s.next_xml_event().await?, Event::Start(b) => assert_eq!(b.local_name().into_inner(), b"starttls")); assert_matches!(s.next_xml_event().await?, Event::Empty(b) => assert_eq!(b.local_name().into_inner(), b"required")); assert_matches!(s.next_xml_event().await?, Event::End(b) => assert_eq!(b.local_name().into_inner(), b"starttls")); assert_matches!(s.next_xml_event().await?, Event::End(b) => assert_eq!(b.local_name().into_inner(), b"features")); s.send(r#""#).await?; assert_matches!(s.next_xml_event().await?, Event::Empty(b) => assert_eq!(b.local_name().into_inner(), b"proceed")); let buffer = s.buffer; tracing::info!("TLS feature negotiation complete"); let connector = TlsConnector::from(Arc::new( ClientConfig::builder() .with_safe_defaults() .with_custom_certificate_verifier(Arc::new(IgnoreCertVerification)) .with_no_client_auth(), )); tracing::info!("Initiating TLS connection..."); let mut stream = connector.connect(ServerName::IpAddress(server.addr.ip()), stream).await?; tracing::info!("TLS connection established"); let mut s = TestScopeTls::new(&mut stream, buffer); s.send(r#""#).await?; s.send(r#""#).await?; assert_matches!(s.next_xml_event().await?, Event::Decl(_) => {}); assert_matches!(s.next_xml_event().await?, Event::Start(b) => assert_eq!(b.local_name().into_inner(), b"stream")); stream.shutdown().await?; // wrap up server.terminate().await?; Ok(()) }