sz-orm-storage 1.2.2

//! # S3 兼容存储后端(真实实现,基于 AWS Signature V4)
//!
//! 本模块提供通用的 S3 兼容 HTTP 客户端,可用于任何兼容 S3 协议的对象存储服务,
//! 包括阿里云 OSS、华为云 OBS、腾讯云 COS、MinIO、Ceph RGW、DigitalOcean Spaces 等。
//!
//! ## 实现要点
//!
//! - 使用 `reqwest`(rustls-tls)发送 HTTP 请求,不依赖 OpenSSL
//! - 使用 `hmac` + `sha2` 手动实现 AWS Signature V4 签名(无 AWS SDK 依赖)
//! - 默认 path-style URL(阿里云/华为/腾讯均推荐 path-style)
//! - 错误处理:HTTP 404 转为 [`StorageError::NotFound`],其他非 2xx 转为对应操作错误
//!
//! ## 启用方式
//!
//! 启用 `s3-compat` feature(或派生 feature `aliyun-oss`/`huawei-obs`/`tencent-cos`):

use crate::error::StorageError;
use crate::signing::{
    derive_signing_key, hex_hmac_sha256, hex_sha256, utc_now_components,
};
use crate::storage::Storage;
use crate::StorageBackend;
use async_trait::async_trait;
use percent_encoding::{utf8_percent_encode, AsciiSet, CONTROLS};

/// AWS Sig V4 签名中需要 percent-encoding 的额外字符集
/// 参考:https://docs.aws.amazon.com/AmazonS3/latest/API/sig-v4-header-based-auth.html
const AWS_ENCODE_SET: &AsciiSet = &CONTROLS
    .add(b' ')
    .add(b'!')
    .add(b'#')
    .add(b'$')
    .add(b'%')
    .add(b'&')
    .add(b'\'')
    .add(b'(')
    .add(b')')
    .add(b'*')
    .add(b'+')
    .add(b',')
    .add(b'/')
    .add(b':')
    .add(b';')
    .add(b'<')
    .add(b'=')
    .add(b'>')
    .add(b'?')
    .add(b'@')
    .add(b'[')
    .add(b']')
    .add(b'{')
    .add(b'}');

/// S3 兼容存储后端
///
/// 通过 AWS Signature V4 签名访问任意 S3 兼容服务。
/// 可被各云 provider(Aliyun/Huawei/Tencent 等)包装复用。
pub struct S3CompatBackend {
    /// 存储桶名称
    pub bucket: String,
    /// 区域(如 us-east-1、cn-north-4)
    pub region: String,
    /// 完整 endpoint URL(含 scheme,如 https://oss-cn-hangzhou.aliyuncs.com)
    pub endpoint: String,
    /// Access Key ID
    pub access_key: String,
    /// Secret Access Key
    pub secret_key: String,
    /// 是否使用 path-style URL(bucket 在 path 中,而非 host 子域)
    pub path_style: bool,
    /// 复用的 HTTP 客户端(连接池)
    client: reqwest::Client,
}

impl S3CompatBackend {
    /// 创建新的 S3 兼容后端实例
    ///
    /// # 参数
    /// - `bucket`: 存储桶名称
    /// - `region`: 区域标识
    /// - `endpoint`: 完整 endpoint(可含或不含 https:// 前缀)
    /// - `access_key` / `secret_key`: 访问密钥
    /// - `path_style`: 是否使用 path-style URL
    pub fn new(
        bucket: impl Into<String>,
        region: impl Into<String>,
        endpoint: impl Into<String>,
        access_key: impl Into<String>,
        secret_key: impl Into<String>,
        path_style: bool,
    ) -> Result<Self, StorageError> {
        // 规范化 endpoint:去除尾部 /,补齐 https:// 前缀
        let endpoint = {
            let e = endpoint.into().trim_end_matches('/').to_string();
            if e.starts_with("http://") || e.starts_with("https://") {
                e
            } else {
                format!("https://{}", e)
            }
        };
        let client = reqwest::Client::builder()
            .build()
            .map_err(|e| StorageError::InvalidConfig(format!("build reqwest client: {}", e)))?;

        Ok(Self {
            bucket: bucket.into(),
            region: region.into(),
            endpoint,
            access_key: access_key.into(),
            secret_key: secret_key.into(),
            path_style,
            client,
        })
    }

    /// 构造对象访问 URL
    ///
    /// path-style: https://endpoint/bucket/key
    /// (virtual-hosted-style 需将 bucket 注入 host,涉及 DNS/TLS SNI,对自建 MinIO 不一定支持,故统一 path-style)
    fn object_url(&self, key: &str) -> String {
        let trimmed_key = key.trim_start_matches('/');
        format!("{}/{}/{}", self.endpoint, self.bucket, trimmed_key)
    }

    /// 生成对象的可读 URL(不含签名,适用于公开桶或内部引用)
    pub fn url_for(&self, key: &str) -> String {
        self.object_url(key)
    }
    /// 计算 AWS Sig V4 签名并返回 Authorization header 值
    ///
    /// 签名流程:
    /// 1. 构造 canonical request(方法、URI、查询串、规范化头、签名头列表、payload 哈希)
    /// 2. 构造 string to sign(算法 + 时间戳 + credential scope + canonical request 哈希)
    /// 3. 派生签名 key(date -> region -> service -> "aws4_request")
    /// 4. HMAC-SHA256(string to sign) 得到签名
    /// 5. 拼装 Authorization header
    fn sign_request(
        &self,
        method: &str,
        url: &reqwest::Url,
        headers: &reqwest::header::HeaderMap,
        payload_hash: &str,
        amz_date: &str,
        date_stamp: &str,
    ) -> Result<String, StorageError> {
        // 1. 解析 host(含端口)
        let host = url
            .host_str()
            .ok_or_else(|| StorageError::InvalidConfig(format!("url missing host: {}", url)))?;
        let host_header = match url.port() {
            Some(p) => format!("{}:{}", host, p),
            None => host.to_string(),
        };

        // 2. canonical URI(对路径分段 percent-encode,保留 /)
        let canonical_uri = canonical_resource_path(url.path());
        // 3. canonical query string(key/value 编码后按字典序排序)
        let pairs: Vec<(String, String)> = url
            .query_pairs()
            .map(|(k, v)| (k.into_owned(), v.into_owned()))
            .collect();
        let canonical_query = canonical_query_string(pairs);

        // 4. 收集需要签名的头(小写键名,排序)
        let mut header_keys: Vec<String> = headers
            .keys()
            .map(|k| k.as_str().to_lowercase())
            .collect();
        // 加入必填隐式头
        for h in ["host", "x-amz-content-sha256", "x-amz-date"] {
            if !header_keys.contains(&h.to_string()) {
                header_keys.push(h.to_string());
            }
        }
        header_keys.sort();

        // 5. 构造 canonical headers(每行 "key:value\n")
        let mut canonical_headers = String::new();
        for k in &header_keys {
            let v = if k == "host" {
                host_header.clone()
            } else {
                headers
                    .get(k.as_str())
                    .map(|hv| hv.to_str().unwrap_or_default().trim().to_string())
                    .unwrap_or_default()
            };
            canonical_headers.push_str(k);
            canonical_headers.push(':');
            canonical_headers.push_str(&v);
            canonical_headers.push('\n');
        }
        let signed_headers = header_keys.join(";");

        // 6. 构造 canonical request
        let canonical_request = format!(
            "{}\n{}\n{}\n{}\n{}\n{}",
            method.to_uppercase(),
            canonical_uri,
            canonical_query,
            canonical_headers,
            signed_headers,
            payload_hash
        );

        // 7. 计算 canonical request 的 SHA256 哈希(十六进制)
        let canonical_request_hash = hex_sha256(canonical_request.as_bytes());

        // 8. credential scope:date/region/service/aws4_request
        let service = "s3";
        let credential_scope = format!("{}/{}/{}/aws4_request", date_stamp, self.region, service);

        // 9. 构造 string to sign
        let string_to_sign = format!(
            "AWS4-HMAC-SHA256\n{}\n{}\n{}",
            amz_date, credential_scope, canonical_request_hash
        );

        // 10. 派生签名 key
        let signing_key = derive_signing_key(&self.secret_key, &date_stamp, &self.region, service);

        // 11. 计算签名
        let signature = hex_hmac_sha256(&signing_key, string_to_sign.as_bytes());

        // 12. 构造 Authorization header
        Ok(format!(
            "AWS4-HMAC-SHA256 Credential={}/{}, SignedHeaders={}, Signature={}",
            self.access_key, credential_scope, signed_headers, signature
        ))
    }
    /// 内部:发送已签名请求并返回响应
    async fn send_signed(
        &self,
        method: reqwest::Method,
        key: &str,
        body: Option<Vec<u8>>,
        extra_headers: Vec<(String, String)>,
    ) -> Result<reqwest::Response, StorageError> {
        let url_str = self.object_url(key);
        let url: reqwest::Url = url_str
            .parse()
            .map_err(|e| StorageError::InvalidConfig(format!("parse url {}: {}", url_str, e)))?;

        // 计算 payload 哈希(空 body 用空串哈希)
        let empty_hash = hex_sha256(b"");
        let payload_hash = body
            .as_ref()
            .map(|b| hex_sha256(b))
            .unwrap_or_else(|| empty_hash.clone());

        // 获取 UTC 时间戳
        let (date_stamp, amz_date) = utc_now_components();

        // 构造请求头
        let mut headers = reqwest::header::HeaderMap::new();
        // x-amz-content-sha256 即使无 body 也必须设置
        headers.insert(
            "x-amz-content-sha256",
            reqwest::header::HeaderValue::from_str(&payload_hash)
                .map_err(|e| StorageError::InvalidConfig(format!("set content-sha256: {}", e)))?,
        );
        headers.insert(
            "x-amz-date",
            reqwest::header::HeaderValue::from_str(&amz_date)
                .map_err(|e| StorageError::InvalidConfig(format!("set amz-date: {}", e)))?,
        );
        // 写入额外头(content-type 等)
        for (k, v) in extra_headers {
            let hname = reqwest::header::HeaderName::from_bytes(k.as_bytes())
                .map_err(|e| StorageError::InvalidConfig(format!("header name {}: {}", k, e)))?;
            let hval = reqwest::header::HeaderValue::from_str(&v)
                .map_err(|e| StorageError::InvalidConfig(format!("header value {}: {}", v, e)))?;
            headers.insert(hname, hval);
        }

        // 计算签名
        let authorization = self.sign_request(
            method.as_str(),
            &url,
            &headers,
            &payload_hash,
            &amz_date,
            &date_stamp,
        )?;

        headers.insert(
            reqwest::header::AUTHORIZATION,
            reqwest::header::HeaderValue::from_str(&authorization)
                .map_err(|e| StorageError::InvalidConfig(format!("set authorization: {}", e)))?,
        );

        // 发送请求
        let req = self
            .client
            .request(method, url)
            .headers(headers)
            .body(body.unwrap_or_default());
        req.send()
            .await
            .map_err(|e| StorageError::Connection(format!("http send: {}", e)))
    }
    /// 上传对象(PUT)
    pub async fn put_object(
        &self,
        key: &str,
        data: &[u8],
        content_type: &str,
    ) -> Result<(), StorageError> {
        let resp = self
            .send_signed(
                reqwest::Method::PUT,
                key,
                Some(data.to_vec()),
                vec![("content-type".to_string(), content_type.to_string())],
            )
            .await?;

        let status = resp.status().as_u16();
        if (200..300).contains(&status) {
            Ok(())
        } else {
            let body = resp.text().await.unwrap_or_default();
            Err(StorageError::Put(format!(
                "s3-compat put status {}: {}",
                status, body
            )))
        }
    }

    /// 下载对象(GET)
    pub async fn get_object(&self, key: &str) -> Result<Vec<u8>, StorageError> {
        let resp = self
            .send_signed(reqwest::Method::GET, key, None, vec![])
            .await?;

        let status = resp.status().as_u16();
        if status == 404 {
            return Err(StorageError::NotFound(key.to_string()));
        }
        if !(200..300).contains(&status) {
            let body = resp.text().await.unwrap_or_default();
            return Err(StorageError::Get(format!(
                "s3-compat get status {}: {}",
                status, body
            )));
        }
        let bytes = resp
            .bytes()
            .await
            .map_err(|e| StorageError::Get(format!("read body: {}", e)))?;
        Ok(bytes.to_vec())
    }

    /// 删除对象(DELETE)
    ///
    /// S3 DELETE 是幂等的:即使对象不存在也返回 204
    pub async fn delete_object(&self, key: &str) -> Result<(), StorageError> {
        let resp = self
            .send_signed(reqwest::Method::DELETE, key, None, vec![])
            .await?;

        let status = resp.status().as_u16();
        if (200..300).contains(&status) {
            Ok(())
        } else {
            let body = resp.text().await.unwrap_or_default();
            Err(StorageError::Delete(format!(
                "s3-compat delete status {}: {}",
                status, body
            )))
        }
    }

    /// 检查对象是否存在(HEAD)
    pub async fn head_object(&self, key: &str) -> Result<bool, StorageError> {
        let resp = self
            .send_signed(reqwest::Method::HEAD, key, None, vec![])
            .await?;

        let status = resp.status().as_u16();
        if (200..300).contains(&status) {
            Ok(true)
        } else if status == 404 {
            Ok(false)
        } else {
            let body = resp.text().await.unwrap_or_default();
            Err(StorageError::Get(format!(
                "s3-compat head status {}: {}",
                status, body
            )))
        }
    }
    /// 列举桶内对象(List Objects V2)
    ///
    /// 返回对象 key 列表(简化实现,最多返回 1000 个)
    pub async fn list_objects(&self, prefix: Option<&str>) -> Result<Vec<String>, StorageError> {
        let url_str = self.object_url("");
        let mut url: reqwest::Url = url_str
            .parse()
            .map_err(|e| StorageError::InvalidConfig(format!("parse url {}: {}", url_str, e)))?;
        // 添加查询参数 list-type=2
        url.query_pairs_mut().append_pair("list-type", "2");
        if let Some(p) = prefix {
            url.query_pairs_mut().append_pair("prefix", p);
        }

        let empty_hash = hex_sha256(b"");
        let (date_stamp, amz_date) = utc_now_components();

        let mut headers = reqwest::header::HeaderMap::new();
        headers.insert(
            "x-amz-content-sha256",
            reqwest::header::HeaderValue::from_str(&empty_hash)
                .map_err(|e| StorageError::InvalidConfig(format!("set content-sha256: {}", e)))?,
        );
        headers.insert(
            "x-amz-date",
            reqwest::header::HeaderValue::from_str(&amz_date)
                .map_err(|e| StorageError::InvalidConfig(format!("set amz-date: {}", e)))?,
        );

        let authorization = self.sign_request(
            "GET",
            &url,
            &headers,
            &empty_hash,
            &amz_date,
            &date_stamp,
        )?;
        headers.insert(
            reqwest::header::AUTHORIZATION,
            reqwest::header::HeaderValue::from_str(&authorization)
                .map_err(|e| StorageError::InvalidConfig(format!("set authorization: {}", e)))?,
        );

        let resp = self
            .client
            .request(reqwest::Method::GET, url)
            .headers(headers)
            .send()
            .await
            .map_err(|e| StorageError::Connection(format!("http send: {}", e)))?;

        let status = resp.status().as_u16();
        if !(200..300).contains(&status) {
            let body = resp.text().await.unwrap_or_default();
            return Err(StorageError::Get(format!(
                "s3-compat list status {}: {}",
                status, body
            )));
        }

        let xml = resp
            .text()
            .await
            .map_err(|e| StorageError::Get(format!("read list body: {}", e)))?;

        // 简易 XML 解析:提取所有 <Key>...</Key>
        let mut keys = Vec::new();
        let marker_start = "<Key>";
        let marker_end = "</Key>";
        let mut cursor = 0;
        while let Some(start_idx) = xml[cursor..].find(marker_start) {
            let abs_start = cursor + start_idx + marker_start.len();
            if let Some(end_idx) = xml[abs_start..].find(marker_end) {
                let abs_end = abs_start + end_idx;
                keys.push(xml[abs_start..abs_end].to_string());
                cursor = abs_end + marker_end.len();
            } else {
                break;
            }
        }
        Ok(keys)
    }
}

#[async_trait]
impl Storage for S3CompatBackend {
    async fn put(
        &self,
        key: &str,
        data: &[u8],
        content_type: &str,
    ) -> Result<String, StorageError> {
        self.put_object(key, data, content_type).await?;
        Ok(self.url_for(key))
    }

    async fn get(&self, key: &str) -> Result<Vec<u8>, StorageError> {
        self.get_object(key).await
    }

    async fn delete(&self, key: &str) -> Result<(), StorageError> {
        self.delete_object(key).await
    }

    async fn exists(&self, key: &str) -> Result<bool, StorageError> {
        self.head_object(key).await
    }
}
// ===================== 工具函数 =====================
//
// 通用签名原语(HMAC-SHA256 / SHA256 / AWS Sig V4 派生 key / UTC 时间)已抽取至
// [`crate::signing`] 模块,此处仅保留 S3 Sig V4 专用的规范化路径/查询串构造逻辑。

/// 规范化资源路径
///
/// AWS Sig V4 要求:URI 路径需进行 URI 编码,但 / 不编码
fn canonical_resource_path(path: &str) -> String {
    if path.is_empty() {
        return "/".to_string();
    }
    let trimmed = path.trim_start_matches('/');
    let encoded: Vec<String> = trimmed
        .split('/')
        .map(|seg| utf8_percent_encode(seg, AWS_ENCODE_SET).to_string())
        .collect();
    let mut result = String::with_capacity(trimmed.len() + 1);
    result.push('/');
    result.push_str(&encoded.join("/"));
    result
}

/// 规范化查询字符串
///
/// 规则:key 与 value 分别 percent-encode,按 key 字典序排序,用 & 连接
fn canonical_query_string(pairs: Vec<(String, String)>) -> String {
    let mut pairs: Vec<(String, String)> = pairs
        .into_iter()
        .map(|(k, v)| {
            (
                utf8_percent_encode(&k, AWS_ENCODE_SET).to_string(),
                utf8_percent_encode(&v, AWS_ENCODE_SET).to_string(),
            )
        })
        .collect();
    pairs.sort();
    pairs
        .iter()
        .map(|(k, v)| format!("{k}={v}"))
        .collect::<Vec<_>>()
        .join("&")
}

// ===================== StorageBackend 实现 =====================
//
// `S3CompatBackend` 已具备 `put_object`/`get_object`/`delete_object`/`head_object`/
// `list_objects` 固有方法,此处通过完全限定语法委托,实现统一的 [`crate::StorageBackend`] trait。
#[async_trait]
impl StorageBackend for S3CompatBackend {
    async fn put_object(
        &self,
        key: &str,
        data: &[u8],
        content_type: &str,
    ) -> Result<(), StorageError> {
        S3CompatBackend::put_object(self, key, data, content_type).await
    }

    async fn get_object(&self, key: &str) -> Result<Vec<u8>, StorageError> {
        S3CompatBackend::get_object(self, key).await
    }

    async fn delete_object(&self, key: &str) -> Result<(), StorageError> {
        S3CompatBackend::delete_object(self, key).await
    }

    async fn head_object(&self, key: &str) -> Result<bool, StorageError> {
        S3CompatBackend::head_object(self, key).await
    }

    async fn list_objects(&self, prefix: Option<&str>) -> Result<Vec<String>, StorageError> {
        S3CompatBackend::list_objects(self, prefix).await
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    // 注:HMAC-SHA256 / SHA256 / AWS 派生 key / UTC 时间等通用签名原语的测试
    // 已统一迁移至 [`crate::signing`] 模块,此处仅保留 S3 Sig V4 专用的规范化逻辑测试。

    #[test]
    fn test_canonical_query_string_sorted() {
        let pairs = vec![
            ("b".to_string(), "2".to_string()),
            ("a".to_string(), "1".to_string()),
            ("c".to_string(), "3".to_string()),
        ];
        let canonical = canonical_query_string(pairs);
        assert_eq!(canonical, "a=1&b=2&c=3");
    }

    #[test]
    fn test_canonical_resource_path_root() {
        assert_eq!(canonical_resource_path(""), "/");
        assert_eq!(canonical_resource_path("/"), "/");
    }

    #[test]
    fn test_canonical_resource_path_simple() {
        let p = canonical_resource_path("/foo/bar");
        assert_eq!(p, "/foo/bar");
    }

    #[test]
    fn test_s3_compat_backend_new() {
        let backend = S3CompatBackend::new(
            "bucket",
            "us-east-1",
            "https://s3.example.com",
            "ak",
            "sk",
            true,
        );
        assert!(backend.is_ok());
        let b = backend.unwrap();
        assert_eq!(b.bucket, "bucket");
        assert_eq!(b.region, "us-east-1");
        assert_eq!(b.endpoint, "https://s3.example.com");
        assert!(b.path_style);
    }

    #[test]
    fn test_s3_compat_backend_endpoint_trailing_slash_stripped() {
        let b = S3CompatBackend::new(
            "bucket",
            "us-east-1",
            "https://s3.example.com/",
            "ak",
            "sk",
            true,
        )
        .unwrap();
        assert_eq!(b.endpoint, "https://s3.example.com");
    }

    #[test]
    fn test_s3_compat_backend_endpoint_adds_scheme() {
        let b = S3CompatBackend::new(
            "bucket",
            "us-east-1",
            "oss-cn-hangzhou.aliyuncs.com",
            "ak",
            "sk",
            true,
        )
        .unwrap();
        assert_eq!(b.endpoint, "https://oss-cn-hangzhou.aliyuncs.com");
    }

    #[test]
    fn test_object_url_path_style() {
        let b = S3CompatBackend::new(
            "my-bucket",
            "us-east-1",
            "https://s3.example.com",
            "ak",
            "sk",
            true,
        )
        .unwrap();
        let url = b.object_url("path/to/file.txt");
        assert_eq!(url, "https://s3.example.com/my-bucket/path/to/file.txt");
    }

    #[tokio::test]
    #[ignore = "requires a real S3-compatible endpoint (e.g. MinIO)"]
    async fn test_s3_compat_real_put_get_delete() {
        let endpoint = std::env::var("S3_COMPAT_TEST_ENDPOINT")
            .unwrap_or_else(|_| "http://localhost:9000".to_string());
        let bucket = std::env::var("S3_COMPAT_TEST_BUCKET")
            .unwrap_or_else(|_| "test-bucket".to_string());
        let region = std::env::var("S3_COMPAT_TEST_REGION")
            .unwrap_or_else(|_| "us-east-1".to_string());
        let access_key = std::env::var("S3_COMPAT_TEST_AK")
            .unwrap_or_else(|_| "minioadmin".to_string());
        let secret_key = std::env::var("S3_COMPAT_TEST_SK")
            .unwrap_or_else(|_| "minioadmin".to_string());

        let backend = S3CompatBackend::new(
            &bucket,
            &region,
            &endpoint,
            &access_key,
            &secret_key,
            true,
        )
        .unwrap();

        let key = format!("s3_compat_test_{}.txt", uuid_simple());
        let data = b"hello s3-compat";

        backend.put_object(&key, data, "text/plain").await.unwrap();
        let fetched = backend.get_object(&key).await.unwrap();
        assert_eq!(fetched, data);

        assert!(backend.head_object(&key).await.unwrap());

        let listed = backend.list_objects(None).await.unwrap();
        assert!(listed.contains(&key));

        backend.delete_object(&key).await.unwrap();
        assert!(!backend.head_object(&key).await.unwrap());
    }

    fn uuid_simple() -> String {
        use std::time::{SystemTime, UNIX_EPOCH};
        let now = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .unwrap_or_default()
            .as_nanos();
        format!("{now:x}")
    }
}