1use super::protocol::{HEADER_LEN, PacketHeader, packet, split_message};
41use rustlavel_core::{Error, Result};
42use std::io;
43use std::pin::Pin;
44use std::sync::Arc;
45use std::task::{Context, Poll, ready};
46use tokio::io::{AsyncRead, AsyncWrite, AsyncWriteExt, ReadBuf};
47use tokio::net::TcpStream;
48
49#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
51pub enum Encryption {
52 Disabled,
55 LoginOnly,
58 #[default]
61 Required,
62}
63
64impl Encryption {
65 pub fn as_byte(self) -> u8 {
67 match self {
68 Encryption::Disabled => super::protocol::encryption::NOT_SUPPORTED,
69 Encryption::LoginOnly => super::protocol::encryption::OFF,
70 Encryption::Required => super::protocol::encryption::ON,
71 }
72 }
73}
74
75#[derive(Debug, Clone, Copy, PartialEq, Eq)]
77pub enum Negotiated {
78 None,
80 LoginOnly,
82 Session,
84}
85
86pub fn negotiate(requested: Encryption, server: u8) -> Result<Negotiated> {
92 use super::protocol::encryption as level;
93
94 Ok(match (requested, server) {
95 (Encryption::Disabled, level::NOT_SUPPORTED) => Negotiated::None,
96 (Encryption::Disabled, level::REQUIRED) | (Encryption::Disabled, level::ON) => {
97 return Err(Error::msg(
98 "the server requires an encrypted connection, but this connection asked for none. \
99 Use the default encryption setting.",
100 ));
101 }
102 (Encryption::Disabled, _) => Negotiated::None,
103
104 (Encryption::LoginOnly, level::NOT_SUPPORTED) => Negotiated::None,
107 (Encryption::Required, level::NOT_SUPPORTED) => {
108 return Err(Error::msg(
109 "this connection requires encryption, but the server reports it cannot encrypt. \
110 Give SQL Server a certificate, or connect with encryption set to login-only.",
111 ));
112 }
113
114 (_, level::ON) | (_, level::REQUIRED) => Negotiated::Session,
117 (Encryption::Required, _) => Negotiated::Session,
118 (Encryption::LoginOnly, _) => Negotiated::LoginOnly,
119 })
120}
121
122pub fn obfuscate_password(password: &str) -> Vec<u8> {
129 password
130 .encode_utf16()
131 .flat_map(u16::to_le_bytes)
132 .map(|byte| byte.rotate_left(4) ^ 0xA5)
134 .collect()
135}
136
137pub fn deobfuscate_password(bytes: &[u8]) -> String {
140 let units: Vec<u16> = bytes
141 .iter()
142 .map(|byte| {
143 let plain = byte ^ 0xA5;
144 plain.rotate_left(4)
145 })
146 .collect::<Vec<u8>>()
147 .chunks_exact(2)
148 .map(|pair| u16::from_le_bytes([pair[0], pair[1]]))
149 .collect();
150
151 String::from_utf16_lossy(&units)
152}
153
154pub struct TdsHandshakeStream {
162 socket: TcpStream,
163 wrapping: bool,
165 packet_size: usize,
166 outgoing: Vec<u8>,
168 pending: Vec<u8>,
170 pending_at: usize,
171 incoming: Vec<u8>,
173 ready: Vec<u8>,
175 ready_at: usize,
176}
177
178impl TdsHandshakeStream {
179 pub fn new(socket: TcpStream, packet_size: usize) -> Self {
180 TdsHandshakeStream {
181 socket,
182 wrapping: true,
183 packet_size,
184 outgoing: Vec::new(),
185 pending: Vec::new(),
186 pending_at: 0,
187 incoming: Vec::new(),
188 ready: Vec::new(),
189 ready_at: 0,
190 }
191 }
192
193 pub fn stop_wrapping(&mut self) {
196 self.wrapping = false;
197 }
198
199 pub fn into_socket(self) -> Result<TcpStream> {
204 if self.ready_at < self.ready.len() || !self.incoming.is_empty() {
205 return Err(Error::Protocol(
206 "the server sent data before encryption was torn down".into(),
207 ));
208 }
209 Ok(self.socket)
210 }
211
212 fn take_packet(&mut self) -> Result<bool> {
214 if self.incoming.len() < HEADER_LEN {
215 return Ok(false);
216 }
217 let header = PacketHeader::parse(&self.incoming)?;
218 let total = header.length as usize;
219 if total < HEADER_LEN {
220 return Err(Error::Protocol("packet length is impossibly small".into()));
221 }
222 if self.incoming.len() < total {
223 return Ok(false);
224 }
225
226 self.ready = self.incoming[HEADER_LEN..total].to_vec();
227 self.ready_at = 0;
228 self.incoming.drain(..total);
229 Ok(true)
230 }
231}
232
233fn protocol_io(error: Error) -> io::Error {
234 io::Error::new(io::ErrorKind::InvalidData, error.to_string())
235}
236
237impl AsyncRead for TdsHandshakeStream {
238 fn poll_read(
239 mut self: Pin<&mut Self>,
240 cx: &mut Context<'_>,
241 buf: &mut ReadBuf<'_>,
242 ) -> Poll<io::Result<()>> {
243 loop {
244 if self.ready_at < self.ready.len() {
245 let take = buf.remaining().min(self.ready.len() - self.ready_at);
246 let at = self.ready_at;
247 buf.put_slice(&self.ready[at..at + take]);
248 self.ready_at += take;
249 return Poll::Ready(Ok(()));
250 }
251
252 if !self.wrapping {
253 if !self.incoming.is_empty() {
256 self.ready = std::mem::take(&mut self.incoming);
257 self.ready_at = 0;
258 continue;
259 }
260 return Pin::new(&mut self.socket).poll_read(cx, buf);
261 }
262
263 if self.take_packet().map_err(protocol_io)? {
264 continue;
265 }
266
267 let mut chunk = [0u8; 8192];
268 let mut incoming = ReadBuf::new(&mut chunk);
269 ready!(Pin::new(&mut self.socket).poll_read(cx, &mut incoming))?;
270
271 let filled = incoming.filled().len();
272 if filled == 0 {
273 return Poll::Ready(Ok(()));
276 }
277 let bytes = incoming.filled().to_vec();
278 self.incoming.extend_from_slice(&bytes);
279 }
280 }
281}
282
283impl AsyncWrite for TdsHandshakeStream {
284 fn poll_write(
285 mut self: Pin<&mut Self>,
286 cx: &mut Context<'_>,
287 buf: &[u8],
288 ) -> Poll<io::Result<usize>> {
289 if !self.wrapping {
290 return Pin::new(&mut self.socket).poll_write(cx, buf);
291 }
292 self.outgoing.extend_from_slice(buf);
295 Poll::Ready(Ok(buf.len()))
296 }
297
298 fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
299 if self.wrapping && !self.outgoing.is_empty() {
300 let flight = std::mem::take(&mut self.outgoing);
301 let packet_size = self.packet_size;
302 for packet in split_message(packet::PRE_LOGIN, &flight, packet_size) {
303 self.pending.extend_from_slice(&packet);
304 }
305 }
306
307 while self.pending_at < self.pending.len() {
308 let this = &mut *self;
309 let written =
310 ready!(Pin::new(&mut this.socket).poll_write(cx, &this.pending[this.pending_at..]))?;
311 if written == 0 {
312 return Poll::Ready(Err(io::ErrorKind::WriteZero.into()));
313 }
314 self.pending_at += written;
315 }
316 self.pending.clear();
317 self.pending_at = 0;
318
319 Pin::new(&mut self.socket).poll_flush(cx)
320 }
321
322 fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
323 Pin::new(&mut self.socket).poll_shutdown(cx)
324 }
325}
326
327pub enum TdsStream {
333 Plain(TcpStream),
334 Tls(Box<tokio_rustls::client::TlsStream<TdsHandshakeStream>>),
335 Closed,
340}
341
342impl TdsStream {
343 pub async fn write_all(&mut self, bytes: &[u8]) -> io::Result<()> {
344 match self {
345 TdsStream::Plain(stream) => stream.write_all(bytes).await,
346 TdsStream::Tls(stream) => stream.write_all(bytes).await,
347 TdsStream::Closed => Err(closed()),
348 }
349 }
350
351 pub async fn flush(&mut self) -> io::Result<()> {
352 match self {
353 TdsStream::Plain(stream) => stream.flush().await,
354 TdsStream::Tls(stream) => stream.flush().await,
355 TdsStream::Closed => Err(closed()),
356 }
357 }
358
359 pub async fn read(&mut self, buffer: &mut [u8]) -> io::Result<usize> {
360 use tokio::io::AsyncReadExt;
361 match self {
362 TdsStream::Plain(stream) => stream.read(buffer).await,
363 TdsStream::Tls(stream) => stream.read(buffer).await,
364 TdsStream::Closed => Err(closed()),
365 }
366 }
367
368 pub async fn shutdown(&mut self) -> io::Result<()> {
369 match self {
370 TdsStream::Plain(stream) => stream.shutdown().await,
371 TdsStream::Tls(stream) => stream.shutdown().await,
372 TdsStream::Closed => Ok(()),
373 }
374 }
375
376 pub fn take(&mut self) -> TdsStream {
378 std::mem::replace(self, TdsStream::Closed)
379 }
380
381 pub fn into_plain(self) -> Result<TdsStream> {
384 match self {
385 TdsStream::Tls(stream) => {
386 let (wrapper, _session) = stream.into_inner();
387 Ok(TdsStream::Plain(wrapper.into_socket()?))
388 }
389 other => Ok(other),
390 }
391 }
392}
393
394fn closed() -> io::Error {
395 io::Error::new(io::ErrorKind::NotConnected, "the TDS connection has no transport")
396}
397
398#[derive(Debug, Clone, Copy)]
400pub struct TlsOptions {
401 pub trust_server_certificate: bool,
411}
412
413impl Default for TlsOptions {
414 fn default() -> Self {
415 TlsOptions { trust_server_certificate: true }
416 }
417}
418
419pub async fn start_tls(
421 socket: TcpStream,
422 host: &str,
423 options: TlsOptions,
424 packet_size: usize,
425) -> Result<tokio_rustls::client::TlsStream<TdsHandshakeStream>> {
426 let connector = tokio_rustls::TlsConnector::from(client_config(options));
427
428 let name = rustls::pki_types::ServerName::try_from(host.to_string())
432 .map_err(|_| Error::msg(format!("`{host}` is not a valid TLS server name")))?;
433
434 let wrapper = TdsHandshakeStream::new(socket, packet_size);
435 let mut stream = connector.connect(name, wrapper).await.map_err(|e| {
436 Error::msg(format!(
437 "the TLS handshake inside SQL Server's pre-login exchange failed: {e}"
438 ))
439 })?;
440
441 stream.get_mut().0.stop_wrapping();
443 Ok(stream)
444}
445
446fn client_config(options: TlsOptions) -> Arc<rustls::ClientConfig> {
460 use std::sync::OnceLock;
461 static TRUSTING: OnceLock<Arc<rustls::ClientConfig>> = OnceLock::new();
462 static VERIFYING: OnceLock<Arc<rustls::ClientConfig>> = OnceLock::new();
463
464 if options.trust_server_certificate {
465 Arc::clone(TRUSTING.get_or_init(|| {
466 let provider = Arc::new(rustls::crypto::aws_lc_rs::default_provider());
467 let verifier = Arc::new(TrustAnyCertificate(Arc::clone(&provider)));
468 Arc::new(
469 builder(provider)
470 .dangerous()
471 .with_custom_certificate_verifier(verifier)
472 .with_no_client_auth(),
473 )
474 }))
475 } else {
476 Arc::clone(VERIFYING.get_or_init(|| {
477 let roots = rustls::RootCertStore { roots: webpki_roots::TLS_SERVER_ROOTS.to_vec() };
480 Arc::new(
481 builder(Arc::new(rustls::crypto::aws_lc_rs::default_provider()))
482 .with_root_certificates(roots)
483 .with_no_client_auth(),
484 )
485 }))
486 }
487}
488
489fn builder(
492 provider: Arc<rustls::crypto::CryptoProvider>,
493) -> rustls::ConfigBuilder<rustls::ClientConfig, rustls::WantsVerifier> {
494 rustls::ClientConfig::builder_with_provider(provider)
495 .with_protocol_versions(&[&rustls::version::TLS12])
496 .expect("TLS 1.2 is enabled by this crate's rustls features")
497}
498
499#[derive(Debug)]
514struct TrustAnyCertificate(Arc<rustls::crypto::CryptoProvider>);
515
516impl rustls::client::danger::ServerCertVerifier for TrustAnyCertificate {
517 fn verify_server_cert(
518 &self,
519 _end_entity: &rustls::pki_types::CertificateDer<'_>,
520 _intermediates: &[rustls::pki_types::CertificateDer<'_>],
521 _server_name: &rustls::pki_types::ServerName<'_>,
522 _ocsp_response: &[u8],
523 _now: rustls::pki_types::UnixTime,
524 ) -> std::result::Result<rustls::client::danger::ServerCertVerified, rustls::Error> {
525 Ok(rustls::client::danger::ServerCertVerified::assertion())
526 }
527
528 fn verify_tls12_signature(
529 &self,
530 _message: &[u8],
531 _cert: &rustls::pki_types::CertificateDer<'_>,
532 _dss: &rustls::DigitallySignedStruct,
533 ) -> std::result::Result<rustls::client::danger::HandshakeSignatureValid, rustls::Error> {
534 Ok(rustls::client::danger::HandshakeSignatureValid::assertion())
535 }
536
537 fn verify_tls13_signature(
538 &self,
539 _message: &[u8],
540 _cert: &rustls::pki_types::CertificateDer<'_>,
541 _dss: &rustls::DigitallySignedStruct,
542 ) -> std::result::Result<rustls::client::danger::HandshakeSignatureValid, rustls::Error> {
543 Ok(rustls::client::danger::HandshakeSignatureValid::assertion())
544 }
545
546 fn supported_verify_schemes(&self) -> Vec<rustls::SignatureScheme> {
547 self.0.signature_verification_algorithms.supported_schemes()
548 }
549}
550
551#[cfg(test)]
552mod tests {
553 use super::*;
554 use super::super::protocol::encryption as level;
555
556 #[test]
557 fn the_password_scheme_swaps_nibbles_then_xors_with_a5() {
558 assert_eq!(obfuscate_password("a"), vec![0xB3, 0xA5]);
562 assert_eq!(obfuscate_password("abc"), vec![0xB3, 0xA5, 0x83, 0xA5, 0x93, 0xA5]);
563
564 assert_eq!(obfuscate_password("é").len(), 2);
566 assert_eq!(obfuscate_password(""), Vec::<u8>::new());
567 }
568
569 #[test]
570 fn obfuscation_is_reversible_which_is_the_whole_point_of_calling_it_that() {
571 for password in ["", "a", "Rustlavel!2026", "pässwörd", "日本語"] {
572 assert_eq!(deobfuscate_password(&obfuscate_password(password)), password);
573 }
574 }
575
576 #[test]
577 fn encryption_choices_map_onto_the_prelogin_option_bytes() {
578 assert_eq!(Encryption::Disabled.as_byte(), level::NOT_SUPPORTED);
579 assert_eq!(Encryption::LoginOnly.as_byte(), level::OFF);
580 assert_eq!(Encryption::Required.as_byte(), level::ON);
581 assert_eq!(Encryption::default(), Encryption::Required);
582 }
583
584 #[test]
585 fn a_server_that_offers_only_login_encryption_gets_login_encryption() {
586 assert_eq!(negotiate(Encryption::LoginOnly, level::OFF).unwrap(), Negotiated::LoginOnly);
587 }
588
589 #[test]
590 fn a_server_that_wants_full_encryption_gets_it_whatever_the_client_preferred() {
591 for server in [level::ON, level::REQUIRED] {
592 assert_eq!(negotiate(Encryption::LoginOnly, server).unwrap(), Negotiated::Session);
593 assert_eq!(negotiate(Encryption::Required, server).unwrap(), Negotiated::Session);
594 }
595 }
596
597 #[test]
598 fn a_server_that_cannot_encrypt_fails_a_connection_that_requires_it() {
599 let error = negotiate(Encryption::Required, level::NOT_SUPPORTED).unwrap_err().to_string();
600 assert!(error.contains("cannot encrypt"), "{error}");
601
602 assert_eq!(negotiate(Encryption::LoginOnly, level::NOT_SUPPORTED).unwrap(), Negotiated::None);
604 }
605
606 #[test]
607 fn refusing_encryption_a_server_requires_is_an_error_not_a_downgrade() {
608 let error = negotiate(Encryption::Disabled, level::REQUIRED).unwrap_err().to_string();
611 assert!(error.contains("requires an encrypted connection"), "{error}");
612
613 assert_eq!(negotiate(Encryption::Disabled, level::NOT_SUPPORTED).unwrap(), Negotiated::None);
614 }
615
616 #[tokio::test]
617 async fn the_handshake_wrapper_frames_a_flight_and_then_gets_out_of_the_way() {
618 use tokio::io::AsyncReadExt;
619
620 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
621 let address = listener.local_addr().unwrap();
622
623 let server = tokio::spawn(async move {
624 let (mut socket, _) = listener.accept().await.unwrap();
625 let mut received = Vec::new();
626 socket.read_buf(&mut received).await.unwrap();
627 let mut more = Vec::new();
629 let _ = tokio::time::timeout(
630 std::time::Duration::from_millis(200),
631 socket.read_buf(&mut more),
632 )
633 .await;
634 received.extend_from_slice(&more);
635 received
636 });
637
638 let mut wrapper =
639 TdsHandshakeStream::new(TcpStream::connect(address).await.unwrap(), 4096);
640 wrapper.write_all(b"handshake").await.unwrap();
641 wrapper.flush().await.unwrap();
642 wrapper.stop_wrapping();
643 wrapper.write_all(b"raw").await.unwrap();
644 wrapper.flush().await.unwrap();
645
646 let received = server.await.unwrap();
647
648 let header = PacketHeader::parse(&received).unwrap();
650 assert_eq!(header.kind, packet::PRE_LOGIN);
651 assert!(header.is_end_of_message());
652 assert_eq!(&received[HEADER_LEN..header.length as usize], b"handshake");
653
654 assert_eq!(&received[header.length as usize..], b"raw");
656 }
657}