feat(gateway): implement S3 read adapter (#487)

Author: beinan

crates/talon-backend/src/s3.rs

@@ -22,7 +22,7 @@ use talon_core::{
     BackendStore, Error, ListPage, ListedObject, ObjectId, ObjectStat, Result, Version,
 };
 
-use crate::http::{HttpClient, HttpRequest, Method};
+use crate::http::{HttpClient, HttpRequest, HttpResponse, HttpStreamResponse, Method};
 
 /// Percent-encode a query value.
 ///
@@ -280,6 +280,77 @@ impl S3Backend {
         req
     }
 
+    fn session_headers(&self) -> Vec<(String, String)> {
+        self.creds
+            .session_token
+            .as_ref()
+            .map(|token| vec![("x-amz-security-token".to_string(), token.clone())])
+            .unwrap_or_default()
+    }
+
+    /// Execute a raw metadata request while forwarding only validated
+    /// conditional headers supplied by a protocol adapter.
+    pub async fn execute_head_raw(
+        &self,
+        obj: &ObjectId,
+        conditions: &[(String, String)],
+    ) -> std::result::Result<HttpResponse, String> {
+        let mut request = self.build_head(obj);
+        request.headers.extend_from_slice(conditions);
+        self.http.execute(self.signed(request)).await
+    }
+
+    /// Execute a whole or ranged GET without buffering its response body.
+    pub async fn execute_get_stream_raw(
+        &self,
+        obj: &ObjectId,
+        range: Option<(u64, u64)>,
+        conditions: &[(String, String)],
+    ) -> std::result::Result<HttpStreamResponse, String> {
+        let mut headers = self.session_headers();
+        if let Some((start, end)) = range {
+            headers.push(("range".into(), format!("bytes={start}-{end}")));
+        }
+        headers.extend_from_slice(conditions);
+        let request = HttpRequest::new(Method::Get, self.object_url(obj), headers);
+        self.http.execute_stream(self.signed(request)).await
+    }
+
+    /// Execute one raw ListObjectsV2 page.
+    pub async fn execute_list_raw(
+        &self,
+        bucket: &str,
+        prefix: &str,
+        delimiter: Option<&str>,
+        continuation_token: Option<&str>,
+        max_keys: u32,
+        encoding_type: Option<&str>,
+    ) -> std::result::Result<HttpResponse, String> {
+        let mut query = format!("list-type=2&max-keys={}", max_keys.min(1000));
+        if !prefix.is_empty() {
+            query.push_str("&prefix=");
+            query.push_str(&encode_query_value(prefix));
+        }
+        if let Some(delimiter) = delimiter {
+            query.push_str("&delimiter=");
+            query.push_str(&encode_query_value(delimiter));
+        }
+        if let Some(token) = continuation_token {
+            query.push_str("&continuation-token=");
+            query.push_str(&encode_query_value(token));
+        }
+        if let Some(encoding_type) = encoding_type {
+            query.push_str("&encoding-type=");
+            query.push_str(&encode_query_value(encoding_type));
+        }
+        let request = HttpRequest::new(
+            Method::Get,
+            format!("{}?{query}", self.bucket_url(bucket)),
+            self.session_headers(),
+        );
+        self.http.execute(self.signed(request)).await
+    }
+
     fn build_streamed_put(
         &self,
         obj: &ObjectId,
@@ -636,6 +707,39 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn raw_list_v2_encodes_and_signs_gateway_parameters() {
+        let http = MockHttp::new(HttpResponse {
+            status: 200,
+            headers: vec![],
+            body: bytes::Bytes::new(),
+        });
+        let mut config = S3Config::aws("us-east-1");
+        config.path_style = true;
+        config.tls = false;
+        config.endpoint = "localhost:4566".into();
+        let s3 = S3Backend::new(config, creds(), http.clone());
+
+        s3.execute_list_raw(
+            "my-bucket",
+            "a/b",
+            Some("/"),
+            Some("next token"),
+            7,
+            Some("url"),
+        )
+        .await
+        .unwrap();
+
+        let request = http.last.lock().unwrap().clone().unwrap();
+        assert_eq!(request.method, Method::Get);
+        assert_eq!(
+            request.url,
+            "http://localhost:4566/my-bucket?list-type=2&max-keys=7&prefix=a%2Fb&delimiter=%2F&continuation-token=next%20token&encoding-type=url"
+        );
+        assert!(request.header("authorization").is_some());
+    }
+
     #[tokio::test]
     async fn fetch_range_returns_body_on_206() {
         let http = MockHttp::new(HttpResponse {

crates/talon-backend/src/sigv4.rs

@@ -77,9 +77,9 @@ fn canonical_query(query: &str) -> String {
         .split('&')
         .filter(|p| !p.is_empty())
         .map(|pair| match pair.split_once('=') {
-            Some((k, v)) => (uri_encode_component(k), uri_encode_component(v)),
+            Some((k, v)) => (uri_encode_query_component(k), uri_encode_query_component(v)),
             // A bare key still needs a trailing `=` in the canonical form.
-            None => (uri_encode_component(pair), String::new()),
+            None => (uri_encode_query_component(pair), String::new()),
         })
         .collect();
     pairs.sort();
@@ -90,14 +90,42 @@ fn canonical_query(query: &str) -> String {
         .join("&")
 }
 
+/// Normalize a URL query component to exactly one layer of AWS encoding.
+///
+/// Request builders must percent-encode values to form a valid URL. SigV4
+/// canonicalization operates on their decoded bytes, so existing escapes are
+/// decoded before applying AWS's RFC 3986 encoding. A literal `+` remains a
+/// plus byte rather than form-urlencoded whitespace.
+fn uri_encode_query_component(value: &str) -> String {
+    let bytes = value.as_bytes();
+    let mut decoded = Vec::with_capacity(bytes.len());
+    let mut index = 0;
+    while index < bytes.len() {
+        if bytes[index] == b'%'
+            && index + 2 < bytes.len()
+            && bytes[index + 1].is_ascii_hexdigit()
+            && bytes[index + 2].is_ascii_hexdigit()
+        {
+            let high = (bytes[index + 1] as char).to_digit(16).unwrap() as u8;
+            let low = (bytes[index + 2] as char).to_digit(16).unwrap() as u8;
+            decoded.push((high << 4) | low);
+            index += 3;
+        } else {
+            decoded.push(bytes[index]);
+            index += 1;
+        }
+    }
+    uri_encode_bytes(&decoded)
+}
+
 /// URI-encode a query key or value.
 ///
 /// Unlike [`uri_encode_path`], `/` is **not** left literal: in a query
 /// component it is a reserved character and must be escaped, which matters for
 /// a listing prefix like `dir/sub`.
-fn uri_encode_component(s: &str) -> String {
-    let mut out = String::with_capacity(s.len());
-    for &b in s.as_bytes() {
+fn uri_encode_bytes(bytes: &[u8]) -> String {
+    let mut out = String::with_capacity(bytes.len());
+    for &b in bytes {
         match b {
             b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'_' | b'.' | b'~' => {
                 out.push(b as char)
@@ -288,6 +316,10 @@ mod tests {
     #[test]
     fn canonical_query_escapes_slashes_in_values() {
         assert_eq!(canonical_query("prefix=dir/sub/"), "prefix=dir%2Fsub%2F");
+        assert_eq!(
+            canonical_query("prefix=dir%2Fsub%2F"),
+            "prefix=dir%2Fsub%2F"
+        );
     }
 
     #[test]
@@ -306,8 +338,8 @@ mod tests {
     }
 
     #[test]
-    fn uri_encode_component_leaves_unreserved_characters_alone() {
-        assert_eq!(uri_encode_component("aZ0-_.~"), "aZ0-_.~");
+    fn uri_encode_query_component_leaves_unreserved_characters_alone() {
+        assert_eq!(uri_encode_query_component("aZ0-_.~"), "aZ0-_.~");
     }
 
     use super::*;

crates/talon-gateway/src/lib.rs

@@ -5,6 +5,7 @@ mod config;
 mod metrics;
 mod model;
 mod runtime;
+pub mod s3;
 
 pub use config::{GatewayConfig, GatewayConfigError, GatewayMode, GatewaySecurity};
 pub use metrics::GatewayMetrics;