xmpp: a basic integration test scenario (#26)

Reviewed-on: lavina/lavina#26

Cargo.lock

@@ -1279,6 +1279,7 @@ name = "projection-xmpp"
 version = "0.0.2-dev"
 dependencies = [
  "anyhow",
+ "assert_matches",
  "derive_more",
  "futures-util",
  "lavina-core",
@@ -1290,6 +1291,7 @@ dependencies = [
  "tokio",
  "tokio-rustls",
  "tracing",
+ "tracing-subscriber",
  "uuid",
 ]
 

Cargo.toml

@@ -28,6 +28,7 @@ tracing = "0.1.37" # logging & tracing api
 prometheus = { version = "0.13.3", default-features = false }
 base64 = "0.21.3"
 lavina-core = { path = "crates/lavina-core" }
+tracing-subscriber = "0.3.16"
 
 [package]
 name = "lavina"
@@ -44,7 +45,7 @@ serde.workspace = true
 serde_json = "1.0.93"
 tokio.workspace = true
 tracing.workspace = true
-tracing-subscriber = "0.3.16"
+tracing-subscriber.workspace = true
 futures-util.workspace = true
 prometheus.workspace = true
 nonempty.workspace = true

crates/projection-xmpp/Cargo.toml

@@ -15,6 +15,10 @@ futures-util.workspace = true
 quick-xml.workspace = true
 proto-xmpp = { path = "../proto-xmpp" }
 uuid = { version = "1.3.0", features = ["v4"] }
-tokio-rustls = "0.24.1"
+tokio-rustls = { version = "0.24.1", features = ["dangerous_configuration"] }
 rustls-pemfile = "1.0.2"
 derive_more.workspace = true
+
+[dev-dependencies]
+tracing-subscriber.workspace = true
+assert_matches = "1.5.0"

crates/projection-xmpp/rustfmt.toml

@@ -0,0 +1,2 @@
+max_width = 120
+chain_width = 120

crates/projection-xmpp/src/lib.rs

@@ -57,13 +57,24 @@ struct Authenticated {
     xmpp_muc_name: Resource,
 }
 
+pub struct RunningServer {
+    pub addr: SocketAddr,
+    terminator: Terminator,
+}
+
+impl RunningServer {
+    pub async fn terminate(self) -> Result<()> {
+        self.terminator.terminate().await
+    }
+}
+
 pub async fn launch(
     config: ServerConfig,
     players: PlayerRegistry,
     rooms: RoomRegistry,
     metrics: MetricsRegistry,
     storage: Storage,
-) -> Result<Terminator> {
+) -> Result<RunningServer> {
     log::info!("Starting XMPP projection");
 
     let certs = certs(&mut SyncBufReader::new(File::open(config.cert)?))?;
@@ -80,6 +91,8 @@ pub async fn launch(
     });
 
     let listener = TcpListener::bind(config.listen_on).await?;
+    let addr = listener.local_addr()?;
+
     let terminator = Terminator::spawn(|mut termination| async move {
         let (stopped_tx, mut stopped_rx) = channel(32);
         let mut actors = HashMap::new();
@@ -138,7 +151,7 @@ pub async fn launch(
         Ok(())
     });
     log::info!("Started XMPP projection");
-    Ok(terminator)
+    Ok(RunningServer { addr, terminator })
 }
 
 async fn handle_socket(
@@ -164,6 +177,7 @@ async fn handle_socket(
         .with_single_cert(vec![config.cert.clone()], config.key.clone())?;
     config.key_log = Arc::new(tokio_rustls::rustls::KeyLogFile::new());
 
+    log::debug!("Accepting TLS connection...");
     let acceptor = TlsAcceptor::from(Arc::new(config));
     let new_stream = acceptor.accept(stream).await?;
     log::debug!("TLS connection established");

crates/projection-xmpp/tests/certs/xmpp.key

@@ -0,0 +1,52 @@
+-----BEGIN PRIVATE KEY-----
+MIIJQgIBADANBgkqhkiG9w0BAQEFAASCCSwwggkoAgEAAoICAQCuViBTGN8tMaQ8
+G8zixL7duElTFCuP6wQhmDsX8ut4V3eEshUpDIIFkCSX17hzfI7duBp1pe7Ket+F
+z5XjbV+ruvxpawvsCgsfGrwXE1vaDVJduy0JyRzLvRSXELWgAbcdllbvBKvGLtY1
+ogm5YJWLbtgQJjutMoLisxn7Xd04fzMQy4aqhy2ZrsxyQSMINuR1Qz/VBzDZi4EH
+Q08rb7GManQfbabbTs1I/GHuAM7PDeb/ou9AZHPASg2fzam5SJhvDYutHmX8wOS3
+b+I+amI6g3N8fJssjx0ryAEL+c+Mbv6mXQhGqh7T++kXtB8h5GoLCOg3yGtaW7o0
+PacAP1UsadDsGN13cWAAsytg1BxqgWk6IqA3Yff5uc2A+TYmX5K5DV46sonovybo
+FI9fdKmL4oCbMIz+Tq+L5vHsdUh5/5S6F2RIKQDJIDYJfE7XVCPyToabQirQsQ/B
+n27L0bCO1hD9cGR+z5td9TPGxrm7GUVGZ/fC58Q4WD/TrVhC3pUq8n7hMzGakg+w
+Ri7FSJTPIZQXiTL/HtPleW+1y6d8Q84UI6Qm39vVueS5YSCjFCiQW2feod+L4sTU
+sE0Rumbvb+saS6cmX7ZBzdgJhP9J5FiAOPqswgCS5w54F5hvfbg+yS4SgeKrvZF3
+dAf+3wW8r039sFN+R/gQowxZcYZwOwIDAQABAoICAEB/p8TmmkcXqxn79RDe1nik
+QiiE+VrtCaG+Nvq0ym5C+fpzgkWmFYKmYgt1aY38gsS/5LYrFk3+KK1ScDNsly0r
+aFA+JPKGgrfWxcjJxj1FmXgJFHAe4lL0WOZM7c1NZSiCoxYaBc00Ldc45F0bwSgN
+cc2Dv6dj3S2vMokfoIVS9hscGW4ExheqJoSM2b+jw2Eo6LhRST7rEGkV+3foAmmf
+RugLwuQ3YtbCXR7XWKwdCh4A84BAydxV6XV6evUMSS0o90isyvG4kcXWFH+gD0hz
+sqnXVfel2RaGD/EU0rczp234VGQEc5RdCk9VOgFphtwfRv7AXQtYjWrfdmYuiD1i
+csiRZxcDgaDbWssBu7oppnbQo0XQ/pynLRJy3Lf9ETIjQ2Xkn6GdoXZY3Phj/XtW
+N7WSQTSEqRfVt7XwWx0wGXlPimIaYJDXmYFU1A/XESrvDgjjwtVrSojipwr5kdYl
+XgCjJVGZyNqAavYzuu7qiH7nXGmEtDQRQlbEQS135ukDkKkv2fIOFbX10iIPgIVM
+Y5dk7Q3gqCWcgcQ/rcR2LLo2H3PdJl0yQnml8rKz2CNpAHp5p4VF66jfsPC0q0qE
+1ad3JAweX042/k0/XPbwIjPCvDBUIeZei3gu8AotgmaRQWgLb7ICgvavjLnJKn/J
+VK2PJHKv/eFvEw6l1T85AoIBAQDVeoTOYe9AdAtsj6oqqjci4WZHJ/keO3pbeh4q
+JKgxzLlyag+8fSGzNkR0RE4f7EF0lnlcRTUpscaes19tmYMzbL84/wmD1SDAjRnG
+OVFkBHIgmXRnyRGQvP0QbBk3mcJKz+8qi7Mvb3YkSdh9lyYs/JtsPDI0XeQrq9J7
+ABIHElu9lCMTtUauljoYZH/pHYjpvk22Ijj6f+0dT5MHHYfsEoxIPB7Ow4JS7buT
+m+O7vlfYZLxSn3OvbdbzGo2IL+AYVDcszSdx6qmHbeu9uMWOvO/YbIX4VyLtzUuc
+MQad7nBOkiKVrTFXb0b+g1dxvUV3+FreZ1K9oFYy82bomcDZAoIBAQDRD7XxRkGO
+OSaWkx2FFMLvntAtO06RpshxtU/rv3vDYwYDulrHc3AaJ+P9rWjb7v15P5OpwRYW
+x1ve8lm8ycKnR6UgD3EYTQkEQkuZ68+ndhVarYCJaelWM9HdhFiWKdPyyrKwcXLr
+blcTZjnq4WC76YtTdSdZfn0KRoSAxuAmVwfWBI4LxaUeMk3TMxbJ4aPUCjkKEw0m
+Jie6S6419d+2aVNXjw4KqaPoSREWlUT8U+iVo1xQp5eg3cOIe2lVukqZTqe7j9Ze
+zP+Gq/RyTk+dvBWcsK/RzhyC73+KD8qAEmdzPAdEwhkxRioTk2PGtU4/K0nDZk3c
+TNlLdeOcqg0zAoIBAEwV9MuSADHaqk+xDJdUP36BE3D9AD8UN9Huvl2K3x+Qte/f
+eWhWuPIkv1UpGycpj1K8ZtjKGd6YbBAYIkTv1+E2OxlXXM7N4XR/VdZei3G4W+ze
+hKyQ71/E2/VEceBtPuBnJ/jj/aNEeLkKUMzCWGrkRYjYE5SyeiZOgSAxsDsxAd2Z
+tL7Ldzu2c1JKT4SIcEnO9+eYXvJ5McumluKMVet/2NvOAbTz3bks3hQIFazOdIS9
+splIF3VJErlml1cYqShCq7+eBxcE6hNIzCK8fj0XfeyHEWCnvd0/tFkg6BjV6NU4
+JHdwWQuur4D60unI6b+OluR5svW+9boHIoB4fFECggEAHeCW6fJWcBLu1toTf+9l
+pIUXzz8IjXw+bTGySEjHUTcXpvS9AIAY50QIKzrbH4NaKjfRzJLRq1O2Z3hPJtHW
+xb1RdfF/AjAQN9GZqFexB4eyqZDeK8U9GZqyRWwilONJbQtW2ix8dfUA8L7NTCoF
+fxVzWewGQZ34FL3bNeQ2KISLlCR2gGwwms4pnSNSAGwE08raN/xdBrSxPMiQDxoi
+bJlE1eCV6yQvToUSsh2HDGCZfrkn+kbZPp4y0ZCBj0TeYGaDRiTaSBYX9pEgkC1s
+52f31rrRhbRlErlTitGS6Ra4Phm4GDV9EDOs07teqQlEM3bmRcybF/7LlyMz8jHD
+TQKCAQEAvjdo48rvwjlj4IKkhkEWP2Qn19bxeQT6lWb9Wtgqar1CIIk9ei74V6xF
+5cyhw9vCXqxvlDM37n0JRHc0EM70aCE7IFL7WxCmmwGXBwT/PeyIJMaaZippd+Ff
+QjBx4z6OSsCnzWnM8YPkRwSSCGVk9EDvxQtwS7BmpqSgilHL1/n1yDKAiiGsvAPI
+uR5WhXzeN/OUHji7Gp5tCxc6Lo4dKqMhQrMFTomUbupw/o4X8TZ0O3hZdzLv9d4Q
+9LM5nI4JOwB4qEUOY9kFQNjvxwPvrTPj8QuIyPIjPQYWZ4Jw8lNCOK3COPV8H+Rb
+kO/SCOEdhp17yWQspky/Uo1RC4lKVA==
+-----END PRIVATE KEY-----

crates/projection-xmpp/tests/certs/xmpp.pem

@@ -0,0 +1,30 @@
+-----BEGIN CERTIFICATE-----
+MIIFDTCCAvWgAwIBAgIUGd9jmau898T/kGdIeiRCcdXueaowDQYJKoZIhvcNAQEL
+BQAwFjEUMBIGA1UEAwwLZXhhbXBsZS5jb20wHhcNMjMxMDA5MTIyMTU2WhcNMjMx
+MTA4MTIyMTU2WjAWMRQwEgYDVQQDDAtleGFtcGxlLmNvbTCCAiIwDQYJKoZIhvcN
+AQEBBQADggIPADCCAgoCggIBAK5WIFMY3y0xpDwbzOLEvt24SVMUK4/rBCGYOxfy
+63hXd4SyFSkMggWQJJfXuHN8jt24GnWl7sp634XPleNtX6u6/GlrC+wKCx8avBcT
+W9oNUl27LQnJHMu9FJcQtaABtx2WVu8Eq8Yu1jWiCblglYtu2BAmO60yguKzGftd
+3Th/MxDLhqqHLZmuzHJBIwg25HVDP9UHMNmLgQdDTytvsYxqdB9tpttOzUj8Ye4A
+zs8N5v+i70Bkc8BKDZ/NqblImG8Ni60eZfzA5Ldv4j5qYjqDc3x8myyPHSvIAQv5
+z4xu/qZdCEaqHtP76Re0HyHkagsI6DfIa1pbujQ9pwA/VSxp0OwY3XdxYACzK2DU
+HGqBaToioDdh9/m5zYD5NiZfkrkNXjqyiei/JugUj190qYvigJswjP5Or4vm8ex1
+SHn/lLoXZEgpAMkgNgl8TtdUI/JOhptCKtCxD8GfbsvRsI7WEP1wZH7Pm131M8bG
+ubsZRUZn98LnxDhYP9OtWELelSryfuEzMZqSD7BGLsVIlM8hlBeJMv8e0+V5b7XL
+p3xDzhQjpCbf29W55LlhIKMUKJBbZ96h34vixNSwTRG6Zu9v6xpLpyZftkHN2AmE
+/0nkWIA4+qzCAJLnDngXmG99uD7JLhKB4qu9kXd0B/7fBbyvTf2wU35H+BCjDFlx
+hnA7AgMBAAGjUzBRMB0GA1UdDgQWBBSFuG4k9lOlZweMeqvBejQic3Me1zAfBgNV
+HSMEGDAWgBSFuG4k9lOlZweMeqvBejQic3Me1zAPBgNVHRMBAf8EBTADAQH/MA0G
+CSqGSIb3DQEBCwUAA4ICAQAjLko9eLBofC9QzF/rm59kOtS6P8fRD0OJKawRdxvP
+5ClQFz9Q/2I7E1jbXL9y8r8iB31hpfnMT/XYxHAmJXyV+X2jHlnKmhzWpfrHx84V
+ZYeFIEFlwJHacPU6mVUUnXLwZIt0VWB3RUT9bgbkXFfkg7Qrccl7Blc0458l8wd2
+mvNgdEGG9Z8VhyaExAHerpOD299llaDTcV+jKLLAboXzDGJsPmMfRu2DWwosfYQJ
+MIKajyylmGOrAk+8fVTVWOPJBE6AmxDpKtZQO04nQq78PohP/CsibalimOf00Wqb
+2x05ssWFP71lmdGilaXCp/mkhaVz/G7ZuDSY9AxCPH4+3pTUbMgsfEmLXRTrUo9c
+B4zz0eDoUiqU9I4TWAHGhnn7b2e6o8ko0baq+PaCSCDK8haL/17CUyMYQBBdhlG6
+sSR2IBSXkxGQZnhfmQdYwc7Y3IgsJJoN0ZfqnyBY+u/ZOBegcZFuhMiDGILJBgFi
+QmgQFTozXKkeEzFdWoRn57lgcBABcMHSucnjekk80uLIWIlL36IyYkPbeI6/I0/l
+ajKftsVEY96hYUXtDBykBpg6gsJXj2gP2FXHW7ngtuI4/mOqI3ltcWh1C83MvM5N
+zMYfTNtxVM1vyvN1M7b7iMeosvmAIvQE3bFk/pJCHAZPfsu0zmeizPHBgCySygLn
+JQ==
+-----END CERTIFICATE-----

crates/projection-xmpp/tests/lib.rs

@@ -0,0 +1,186 @@
+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<ReadHalf<'_>>, buf: &mut Vec<u8>) -> Result<usize> {
+    let mut size = 0;
+    let res = reader.read_until(b'\n', buf).await?;
+    size += res;
+    return Ok(size);
+}
+
+struct TestScope<'a> {
+    reader: NsReader<BufReader<ReadHalf<'a>>>,
+    writer: WriteHalf<'a>,
+    buffer: Vec<u8>,
+}
+
+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.flush().await?;
+        Ok(())
+    }
+
+    async fn next_xml_event(&mut self) -> Result<Event<'_>> {
+        self.buffer.clear();
+        let event = self.reader.read_event_into_async(&mut self.buffer).await?;
+        Ok(event)
+    }
+
+    async fn read<T: FromXml>(&mut self) -> Result<T> {
+        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, anyhow::Error>> = 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<BufReader<GenericReadHalf<&'a mut TlsStream<TcpStream>>>>,
+    writer: GenericWriteHalf<&'a mut TlsStream<TcpStream>>,
+    buffer: Vec<u8>,
+    pub timeout: Duration,
+}
+
+impl<'a> TestScopeTls<'a> {
+    fn new(stream: &'a mut TlsStream<TcpStream>, buffer: Vec<u8>) -> 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.flush().await?;
+        Ok(())
+    }
+
+    async fn next_xml_event(&mut self) -> Result<Event<'_>> {
+        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<Item = &[u8]>,
+        _ocsp_response: &[u8],
+        _now: std::time::SystemTime,
+    ) -> std::result::Result<tokio_rustls::rustls::client::ServerCertVerified, tokio_rustls::rustls::Error> {
+        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#"<?xml version="1.0"?>"#).await?;
+    s.send(r#"<stream:stream xmlns:stream="http://etherx.jabber.org/streams" to="127.0.0.1" xml:lang="en" xmlns:xml="http://www.w3.org/XML/1998/namespace" xmlns="jabber:client" version="1.0">"#).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#"<starttls/>"#).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#"<?xml version="1.0"?>"#).await?;
+    s.send(r#"<stream:stream xmlns:stream="http://etherx.jabber.org/streams" to="127.0.0.1" xml:lang="en" xmlns:xml="http://www.w3.org/XML/1998/namespace" xmlns="jabber:client" version="1.0">"#).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(())
+}