[External]Add S3 blob storage with cashier billing to ic-gateway

Cargo.toml

@@ -19,8 +19,12 @@ ahash = "0.8.11"
 anyhow = "1.0.93"
 arc-swap = "1.7.1"
 async-trait = "0.1.83"
+aws-config = "1.8"
+aws-sdk-s3 = "1.120"
+aws-smithy-http-client = { version = "1.1", features = ["rustls-ring"] }
+aws-smithy-runtime-api = { version = "1.10", features = ["client"] }
 axum = { version = "0.8.1", features = ["macros"] }
-axum-extra = "0.10.0"
+axum-extra = { version = "0.10.0", features = ["typed-header"] }
 bytes = "1.10.0"
 candid = "0.10.10"
 clap = { version = "4.5.20", features = ["derive", "string", "env"] }
@@ -50,6 +54,7 @@ ic-bn-lib = { version = "0.1.18", features = [
     "clients-hyper",
 ] }
 ic-bn-lib-common = "0.1"
+ic-certificate-verification = "3.1"
 ic-custom-domains-backend = "0.1"
 ic-custom-domains-base = "0.1"
 ic-http-certification = { version = "3.1.0", optional = true }
@@ -87,6 +92,7 @@ tokio = { version = "1.45.0", features = ["full", "tracing"] }
 tokio-util = { version = "0.7.12", features = ["full"] }
 tower = { version = "0.5.1", features = ["limit"] }
 tower-http = { version = "0.6.1", features = ["cors", "compression-full"] }
+thiserror = "2.0.18"
 tracing = "0.1.40"
 tracing-core = "0.1.32"
 tracing-serde = "0.2.0"
@@ -100,6 +106,8 @@ url = "2.5.3"
 # Read https://github.com/uuid-rs/uuid/releases/tag/1.13.0
 uuid = { version = "=1.12.1", features = ["v7"] }
 woothee = "0.13.0"
+async-stream = "0.3.6"
+ic-certification = "3"
 
 [dev-dependencies]
 criterion = { version = "0.5", features = ["html_reports", "async_tokio"] }

src/api.rs

@@ -66,11 +66,7 @@ pub async fn log_handler(
             .into_response();
     };
     // Maintain hickory_proto::dnssec=error filter when changing log level
-    let env_filter = EnvFilter::new(format!(
-        "{},{}",
-        log_level,
-        crate::log::LOG_LEVEL_OVERRIDES
-    ));
+    let env_filter = EnvFilter::new(format!("{},{}", log_level, crate::log::LOG_LEVEL_OVERRIDES));
     let _ = state.log_handle.modify(|f| *f = env_filter);
 
     "Ok\n".into_response()

src/cli.rs

@@ -17,9 +17,11 @@ use ic_bn_lib_common::{
 };
 use reqwest::Url;
 
+use candid::Principal;
+
 use crate::{
     core::{AUTHOR_NAME, SERVICE_NAME},
-    routing::{RequestType, domain::CanisterAlias},
+    routing::{RequestType, domain::CanisterAlias, storage::S3Flavor},
 };
 
 /// Clap does not support prefixes due to macro limitations.
@@ -91,6 +93,9 @@ pub struct Cli {
     #[command(flatten, next_help_heading = "Cache")]
     pub cache: CacheConfig,
 
+    #[command(flatten)]
+    pub blob_storage: BlobStorage,
+
     #[command(flatten, next_help_heading = "Shedding System")]
     pub shed_system: ShedSystemCli,
 
@@ -531,6 +536,69 @@ pub struct CacheConfig {
     pub cache_xfetch_beta: f64,
 }
 
+/// Blob-storage feature config: cashier billing + S3 backend.
+/// Both groups below are flattened at the top level; `cli.blob_storage.cashier.*`
+/// and `cli.blob_storage.s3.*` on the Rust side, flag names unchanged.
+#[derive(Args)]
+pub struct BlobStorage {
+    #[command(flatten, next_help_heading = "Blob Storage — Cashier")]
+    pub cashier: CashierConfig,
+
+    #[command(flatten, next_help_heading = "Blob Storage — S3")]
+    pub s3: S3Storage,
+}
+
+#[derive(Args)]
+pub struct CashierConfig {
+    /// Canister ID of the cashier backend.
+    /// When set, the gateway will connect to this canister for billing and budget checks.
+    #[clap(env, long)]
+    pub cashier_canister_id: Option<Principal>,
+
+    /// How frequently to report accumulated usage counters to the cashier canister
+    #[clap(env, long, default_value = "10s", value_parser = parse_duration)]
+    pub cashier_usage_report_interval: Duration,
+
+    /// Comma-separated list of hosts allowed to call DELETE /owner.
+    /// If unset, owner deletion is unrestricted.
+    #[clap(env = "ALLOW_DELETE_OWNER_FROM_HOST", long)]
+    pub allow_delete_owner_from_host: Option<String>,
+}
+
+#[derive(Args)]
+pub struct S3Storage {
+    /// S3-compatible endpoint URL (e.g. http://localhost:9000 for MinIO).
+    /// When set together with other S3_* options, enables S3 storage backend.
+    #[clap(env, long)]
+    pub s3_endpoint: Option<String>,
+
+    /// S3 access key
+    #[clap(env, long, default_value = "root-user")]
+    pub s3_access_key: String,
+
+    /// S3 secret key
+    #[clap(env, long, default_value = "password")]
+    pub s3_secret_key: String,
+
+    /// S3 bucket name (used as the default / fallback bucket for health checks)
+    #[clap(env, long, default_value = "ic-gateway-storage")]
+    pub s3_bucket: String,
+
+    /// S3 region
+    #[clap(env, long, default_value = "us-east-1")]
+    pub s3_region: String,
+
+    /// S3 session token (for temporary credentials, e.g. Okta-based AWS access)
+    #[clap(env, long)]
+    pub s3_session_token: Option<String>,
+
+    /// Which S3 flavor we're talking to. Drives feature selection
+    /// (e.g. whether to request `INTELLIGENT_TIERING` on uploads) without
+    /// probing the backend at startup. Defaults to `minio` for local dev.
+    #[clap(env, long, value_enum, default_value_t = S3Flavor::Minio)]
+    pub s3_flavor: S3Flavor,
+}
+
 #[derive(Args)]
 pub struct Cors {
     /// Default value for Access-Control-Allow-Origin header

src/core.rs

@@ -31,6 +31,10 @@ use crate::{
     cli::Cli,
     metrics,
     routing::ic::subnets_info::SubnetsInfoFetcher,
+    routing::storage::{
+        AWSBucket, BucketLike, CashierClient, CashierConnector, IngressAuth, IngressAuthImpl,
+        S3Config, StorageState,
+    },
     routing::{
         self,
         ic::{
@@ -254,7 +258,9 @@ pub async fn main(
 
     let root_subnet_id = principal!(MAINNET_ROOT_SUBNET_ID);
 
-    let fetcher = Arc::new(SubnetsInfoFetcher::new(Arc::new(agent), root_subnet_id));
+    let agent = Arc::new(agent);
+
+    let fetcher = Arc::new(SubnetsInfoFetcher::new(agent.clone(), root_subnet_id));
     let subnets_info = fetcher.info.clone();
 
     health_manager.add(fetcher.clone());
@@ -264,6 +270,89 @@ pub async fn main(
         cli.domain.subnets_info_poll_interval,
     );
 
+    // Setup Cashier client + billing connector
+    let cashier_connector = if let Some(canister_id) = cli.blob_storage.cashier.cashier_canister_id
+    {
+        let cashier_client = Arc::new(CashierClient::new(agent.clone(), canister_id));
+        warn!("Cashier client configured for canister {canister_id}");
+
+        match CashierConnector::new(cashier_client, Some(HOSTNAME.get().unwrap().clone())).await {
+            Ok(connector) => {
+                let connector = Arc::new(connector);
+                warn!("CashierConnector initialized (billing enabled)");
+
+                health_manager.add(connector.clone());
+                tasks.add_interval(
+                    "cashier_usage_reporter",
+                    connector.clone(),
+                    cli.blob_storage.cashier.cashier_usage_report_interval,
+                );
+
+                Some(connector)
+            }
+            Err(e) => {
+                warn!("CashierConnector init failed (non-fatal, billing disabled): {e:#}");
+                None
+            }
+        }
+    } else {
+        None
+    };
+
+    // Setup S3 storage backend (single bucket)
+    let s3_bucket = if let Some(ref endpoint) = cli.blob_storage.s3.s3_endpoint {
+        let flavor = cli.blob_storage.s3.s3_flavor;
+        let s3_config = S3Config {
+            endpoint: endpoint.clone(),
+            access_key: cli.blob_storage.s3.s3_access_key.clone(),
+            secret_key: cli.blob_storage.s3.s3_secret_key.clone(),
+            bucket_name: cli.blob_storage.s3.s3_bucket.clone(),
+            region: cli.blob_storage.s3.s3_region.clone(),
+            session_token: cli.blob_storage.s3.s3_session_token.clone(),
+            flavor,
+        };
+        warn!(
+            endpoint = %endpoint,
+            bucket = %s3_config.bucket_name,
+            region = %s3_config.region,
+            flavor = ?flavor,
+            "Initializing S3 storage backend"
+        );
+
+        match AWSBucket::new(s3_config, Arc::new(dns_resolver.clone())).await {
+            Ok(bucket) => {
+                warn!(
+                    bucket = %cli.blob_storage.s3.s3_bucket,
+                    "S3 bucket ready"
+                );
+                Some(Arc::new(bucket) as Arc<dyn BucketLike>)
+            }
+            Err(e) => {
+                warn!("S3 bucket initialization failed (non-fatal): {e}");
+                None
+            }
+        }
+    } else {
+        None
+    };
+
+    // Assemble storage state (enabled iff both S3 bucket and cashier connector
+    // are configured). Ingress auth is only constructed when storage is active.
+    let storage_state = s3_bucket.zip(cashier_connector).map(|(bucket, connector)| {
+        let ingress_auth: Arc<dyn IngressAuth> =
+            Arc::new(IngressAuthImpl::new(agent.read_root_key()));
+        StorageState {
+            connector,
+            bucket,
+            ingress_auth,
+            allowed_delete_owner_hosts: cli
+                .blob_storage
+                .cashier
+                .allow_delete_owner_from_host
+                .clone(),
+        }
+    });
+
     // Setup WAF
     let waf_layer = if cli.waf.waf_enable {
         let v = WafLayer::new_from_cli(&cli.waf, Some(http_client.clone()))
@@ -292,6 +381,7 @@ pub async fn main(
         waf_layer,
         custom_domains_router,
         subnets_info,
+        storage_state,
     )
     .await
     .context("unable to setup Axum router")?;