test(storage): cover GCS HTTPS OpenDAL TLS

Author: palkakrzysiek

quickwit/Cargo.lock

@@ -62,7 +62,10 @@ reqwest = { workspace = true, optional = true }
 http = { workspace = true }
 mockall = { workspace = true }
 proptest = { workspace = true }
+reqwest-013 = { package = "reqwest", version = "0.13", default-features = false }
+rustls = { workspace = true }
 tokio = { workspace = true }
+tokio-rustls = { workspace = true }
 tracing-subscriber = { workspace = true }
 
 aws-sdk-s3 = { workspace = true }

quickwit/quickwit-storage/Cargo.toml

@@ -113,10 +113,55 @@ fn parse_google_uri(uri: &Uri) -> Option<(String, PathBuf)> {
 
 #[cfg(test)]
 mod tests {
+    use std::sync::Arc;
+
+    use base64::Engine;
+    use opendal::Operator;
+    use opendal::layers::HttpClientLayer;
+    use opendal::raw::HttpClient;
     use quickwit_common::uri::Uri;
+    use rustls::pki_types::{CertificateDer, PrivateKeyDer, PrivateSec1KeyDer};
+    use tokio::io::{AsyncReadExt, AsyncWriteExt};
+    use tokio::net::TcpListener;
+    use tokio::task::JoinHandle;
 
     use super::parse_google_uri;
 
+    // Test-only localhost certificate chain. The client below trusts only this
+    // embedded CA, so the test never depends on the host root store.
+    const TEST_CA_CERT_DER_BASE64: &str = concat!(
+        "MIIBizCCATGgAwIBAgIURn3X2TGz13Wv1/p3x8S3y1ALQwkwCgYIKoZIzj0EAwIw",
+        "GzEZMBcGA1UEAwwQUXVpY2t3aXQgVGVzdCBDQTAeFw0yNjA1MzAxMTE4NDJaFw0z",
+        "NjA1MjcxMTE4NDJaMBsxGTAXBgNVBAMMEFF1aWNrd2l0IFRlc3QgQ0EwWTATBgcq",
+        "hkjOPQIBBggqhkjOPQMBBwNCAATyMe2ARkjYkk+QnWZA8T+A99xLcHhMRdcMBElb",
+        "NFYf2nccWkaaqEnUGOIe6Y6ODehYX5dpEWpwkXQ20p52mtDLo1MwUTAdBgNVHQ4E",
+        "FgQUgwSEbLYdLqTp4uEWN3CqY4LkHgUwHwYDVR0jBBgwFoAUgwSEbLYdLqTp4uEW",
+        "N3CqY4LkHgUwDwYDVR0TAQH/BAUwAwEB/zAKBggqhkjOPQQDAgNIADBFAiAvxQlF",
+        "s1tG0Qs/rY9WMcKdvCiOyW044KePy4t9jzkhsgIhALBerDKsWc80bgBYWetpIZVS",
+        "FdrixGuySBCdAG9w0rWn",
+    );
+
+    const TEST_SERVER_CERT_DER_BASE64: &str = concat!(
+        "MIIBuDCCAV6gAwIBAgIUIwRonRLHFnBmsr8U5Ytq9LFW6ZMwCgYIKoZIzj0EAwIw",
+        "GzEZMBcGA1UEAwwQUXVpY2t3aXQgVGVzdCBDQTAeFw0yNjA1MzAxMTE4NDJaFw0z",
+        "NjA1MjcxMTE4NDJaMBQxEjAQBgNVBAMMCWxvY2FsaG9zdDBZMBMGByqGSM49AgEG",
+        "CCqGSM49AwEHA0IABFqZwG11V1JJijKkWXhVkFOoq5/VAjkjejQb0BcQRadBQTN2",
+        "yziIrxFWSgfnekyIa3j7Zfa4WIPL3LLfrjBg6xOjgYYwgYMwFAYDVR0RBA0wC4IJ",
+        "bG9jYWxob3N0MAkGA1UdEwQCMAAwCwYDVR0PBAQDAgWgMBMGA1UdJQQMMAoGCCsG",
+        "AQUFBwMBMB0GA1UdDgQWBBQecwDdu1wYyfkDKm8RsEBUzl3eijAfBgNVHSMEGDAW",
+        "gBSDBIRsth0upOni4RY3cKpjguQeBTAKBggqhkjOPQQDAgNIADBFAiBhbBIAKEID",
+        "6F0vribdLyPkJbMoNNOSK79hdrpdnYyTZwIhALwHMlmusqbK2lzH1l6bwbpldrfb",
+        "xiQgxb9SrOgsPmwz",
+    );
+
+    const TEST_SERVER_KEY_DER_BASE64: &str = concat!(
+        "MHcCAQEEIB39TDCC+qOWh5pRcqaGvaAY2Zx9hYWgREB/QoFehxH2oAoGCCqGSM49",
+        "AwEHoUQDQgAEWpnAbXVXUkmKMqRZeFWQU6irn9UCOSN6NBvQFxBFp0FBM3bLOIiv",
+        "EVZKB+d6TIhrePtl9rhYg8vcst+uMGDrEw==",
+    );
+
+    type LocalHttpsGcsServer = (String, JoinHandle<anyhow::Result<()>>);
+
     #[test]
     fn test_parse_google_uri() {
         assert!(parse_google_uri(&Uri::for_test("gs://")).is_none());
@@ -134,4 +179,96 @@ mod tests {
         assert_eq!(bucket, "test-bucket");
         assert_eq!(prefix.to_str().unwrap(), "indexes");
     }
+
+    #[tokio::test]
+    #[cfg_attr(not(feature = "ci-test"), ignore)]
+    async fn test_gcs_https_endpoint_can_read_from_local_server() -> anyhow::Result<()> {
+        let (endpoint, server_task) = start_local_https_gcs_server().await?;
+        let ca_cert_der = decode_test_der(TEST_CA_CERT_DER_BASE64)?;
+        let ca_cert = reqwest_013::Certificate::from_der(&ca_cert_der)?;
+        let reqwest_client = reqwest_013::Client::builder()
+            .no_proxy()
+            .tls_certs_only([ca_cert])
+            .build()?;
+
+        let cfg = opendal::services::Gcs::default()
+            .bucket("quickwit-test-bucket")
+            .endpoint(&endpoint)
+            .allow_anonymous()
+            .disable_config_load()
+            .disable_vm_metadata();
+        let op = Operator::new(cfg)?
+            .layer(HttpClientLayer::new(HttpClient::with(reqwest_client)))
+            .finish();
+
+        let bytes = op.read_with("hello.txt").range(0..2).await?.to_bytes();
+        assert_eq!(bytes.as_ref(), b"ok");
+        server_task.await??;
+        Ok(())
+    }
+
+    async fn start_local_https_gcs_server() -> anyhow::Result<LocalHttpsGcsServer> {
+        let cert_chain = vec![CertificateDer::from(decode_test_der(
+            TEST_SERVER_CERT_DER_BASE64,
+        )?)];
+        let private_key = PrivateKeyDer::Sec1(PrivateSec1KeyDer::from(decode_test_der(
+            TEST_SERVER_KEY_DER_BASE64,
+        )?));
+        let tls_config = rustls::ServerConfig::builder()
+            .with_no_client_auth()
+            .with_single_cert(cert_chain, private_key)?;
+        let tls_acceptor = tokio_rustls::TlsAcceptor::from(Arc::new(tls_config));
+        let listener = TcpListener::bind(("127.0.0.1", 0)).await?;
+        let endpoint = format!("https://localhost:{}", listener.local_addr()?.port());
+
+        let server_task = tokio::spawn(async move {
+            let (stream, _) = listener.accept().await?;
+            let mut stream = tls_acceptor.accept(stream).await?;
+            let mut request = Vec::new();
+            let mut buffer = [0u8; 1024];
+            loop {
+                let bytes_read = stream.read(&mut buffer).await?;
+                if bytes_read == 0 {
+                    break;
+                }
+                request.extend_from_slice(&buffer[..bytes_read]);
+                if request.windows(4).any(|window| window == b"\r\n\r\n") {
+                    break;
+                }
+            }
+            let request = String::from_utf8_lossy(&request);
+            assert!(
+                request.starts_with(
+                    "GET /storage/v1/b/quickwit-test-bucket/o/hello.txt?alt=media HTTP/"
+                ),
+                "unexpected GCS request line: {request}"
+            );
+            assert!(
+                request
+                    .to_ascii_lowercase()
+                    .contains("\r\nrange: bytes=0-1\r\n"),
+                "expected range read request header: {request}"
+            );
+
+            stream
+                .write_all(
+                    b"HTTP/1.1 206 Partial Content\r\n\
+                    Content-Length: 2\r\n\
+                    Content-Range: bytes 0-1/2\r\n\
+                    Accept-Ranges: bytes\r\n\
+                    Connection: close\r\n\
+                    \r\n\
+                    ok",
+                )
+                .await?;
+            stream.shutdown().await?;
+            Ok(())
+        });
+
+        Ok((endpoint, server_task))
+    }
+
+    fn decode_test_der(base64_der: &str) -> anyhow::Result<Vec<u8>> {
+        Ok(base64::engine::general_purpose::STANDARD.decode(base64_der)?)
+    }
 }