tryaudex_core/

gcp.rs

use std::sync::Arc;
use std::time::Duration;

use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};

use crate::error::{AvError, Result};

/// Temporary GCP credentials (short-lived OAuth2 access token).
#[derive(Clone, Serialize, Deserialize)]
pub struct GcpTempCredentials {
    #[serde(skip_serializing)]
    pub access_token: String,
    pub expires_at: DateTime<Utc>,
}

impl std::fmt::Debug for GcpTempCredentials {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("GcpTempCredentials")
            .field("access_token", &"[REDACTED]")
            .field("expires_at", &self.expires_at)
            .finish()
    }
}

/// Issues short-lived GCP credentials via Service Account impersonation.
/// Stores the auth provider so tokens are refreshed on each call,
/// avoiding stale token errors in long-running processes (e.g., server mode).
pub struct GcpCredentialIssuer {
    auth_provider: Arc<dyn gcp_auth::TokenProvider>,
    http: reqwest::Client,
}

#[derive(Deserialize)]
struct GenerateAccessTokenResponse {
    #[serde(rename = "accessToken")]
    access_token: String,
    #[serde(rename = "expireTime")]
    expire_time: String,
}

impl GcpCredentialIssuer {
    /// Create a new issuer using Application Default Credentials.
    /// Stores the auth provider so tokens are refreshed per-call.
    pub async fn new() -> Result<Self> {
        let auth = gcp_auth::provider().await.map_err(|e| {
            AvError::Gcp(format!(
                "Failed to get GCP credentials. Run `gcloud auth application-default login` first. Error: {}",
                e
            ))
        })?;

        // Verify credentials work at init time
        auth.token(&["https://www.googleapis.com/auth/cloud-platform"])
            .await
            .map_err(|e| AvError::Gcp(format!("Failed to get access token: {}", e)))?;

        let http = reqwest::Client::builder()
            .timeout(Duration::from_secs(30))
            .connect_timeout(Duration::from_secs(10))
            .build()
            .map_err(|e| AvError::Gcp(format!("HTTP client error: {}", e)))?;

        Ok(Self {
            auth_provider: auth,
            http,
        })
    }

    /// Get a fresh source token for API calls.
    async fn source_token(&self) -> Result<String> {
        let token = self
            .auth_provider
            .token(&["https://www.googleapis.com/auth/cloud-platform"])
            .await
            .map_err(|e| AvError::Gcp(format!("Failed to refresh source token: {}", e)))?;
        Ok(token.as_str().to_string())
    }

    /// Generate a short-lived access token by impersonating a service account.
    ///
    /// Calls the IAM Credentials API `generateAccessToken` endpoint.
    /// The source credentials (from ADC) must have `iam.serviceAccounts.getAccessToken`
    /// permission on the target service account.
    /// Generate a short-lived access token. Set `extended_lifetime` to true
    /// if the target project has the org policy
    /// `constraints/iam.allowServiceAccountCredentialLifetimeExtension` enabled,
    /// which raises the cap from 1h to 12h.
    pub async fn issue(
        &self,
        service_account: &str,
        ttl: Duration,
        extended_lifetime: bool,
    ) -> Result<GcpTempCredentials> {
        // Validate service account format to prevent URL path injection.
        // Reject double-@, control characters, and non-email formats.
        let is_valid_sa = {
            let parts: Vec<&str> = service_account.split('@').collect();
            parts.len() == 2
                && !parts[0].is_empty()
                && parts[1].ends_with(".iam.gserviceaccount.com")
                && service_account
                    .bytes()
                    .all(|b| b.is_ascii_alphanumeric() || b"@.-_".contains(&b))
        };
        if !is_valid_sa {
            return Err(AvError::Gcp(format!(
                "Invalid service account email '{}': expected format 'name@project.iam.gserviceaccount.com'",
                service_account
            )));
        }

        let max_ttl = if extended_lifetime { 43200 } else { 3600 };
        let ttl_secs = ttl.as_secs().min(max_ttl);
        if ttl.as_secs() > max_ttl {
            tracing::warn!(
                requested = ttl.as_secs(),
                actual = ttl_secs,
                max = max_ttl,
                "GCP access token TTL clamped to maximum allowed"
            );
        }

        let encoded_sa = urlencoding::encode(service_account);
        let url = format!(
            "https://iamcredentials.googleapis.com/v1/projects/-/serviceAccounts/{}:generateAccessToken",
            encoded_sa
        );

        let oauth_scopes = narrow_oauth_scopes(service_account);
        let body = serde_json::json!({
            "scope": oauth_scopes,
            "lifetime": format!("{}s", ttl_secs),
        });

        let fresh_token = self.source_token().await?;

        let client = &self.http;
        let resp = client
            .post(&url)
            .bearer_auth(&fresh_token)
            .json(&body)
            .send()
            .await
            .map_err(|e| AvError::Gcp(format!("IAM Credentials API call failed: {}", e)))?;

        if !resp.status().is_success() {
            let status = resp.status();
            let text = resp.text().await.unwrap_or_default();
            let sanitized = sanitize_gcp_error(&text);
            return Err(AvError::Gcp(format!(
                "IAM Credentials API error ({}): {}",
                status, sanitized
            )));
        }

        let result: GenerateAccessTokenResponse = resp.json().await.map_err(|e| {
            AvError::Gcp(format!("Failed to parse IAM Credentials response: {}", e))
        })?;

        let expires_at = DateTime::parse_from_rfc3339(&result.expire_time)
            .map(|dt| dt.with_timezone(&Utc))
            .map_err(|e| {
                AvError::Gcp(format!(
                    "Failed to parse token expiry '{}': {}",
                    result.expire_time, e
                ))
            })?;

        Ok(GcpTempCredentials {
            access_token: result.access_token,
            expires_at,
        })
    }

    /// Downscope an access token using GCP Credential Access Boundaries (CAB).
    ///
    /// Exchanges the source token for a downscoped token via the STS endpoint.
    /// CAB supports a subset of GCP services (primarily Cloud Storage). For
    /// unsupported permissions, this returns an error and the caller should
    /// fall back to advisory-only mode with a warning.
    ///
    /// Returns the downscoped credentials plus the list of requested actions
    /// that could not be enforced via CAB (R6-M18). When the skipped list is
    /// non-empty, callers MUST treat the token as having only partial scoping
    /// — the returned CAB rules cover Cloud Storage only, and any non-Storage
    /// action in the `--allow` list is effectively denied at runtime but the
    /// operator's intent is not honoured by the token boundary.
    ///
    /// See: https://cloud.google.com/iam/docs/downscoping-short-lived-credentials
    pub async fn downscope_token(
        &self,
        source_token: &str,
        actions: &[String],
        resources: &[&str],
        source_expires_at: Option<DateTime<Utc>>,
    ) -> Result<(GcpTempCredentials, Vec<String>)> {
        let (rules, skipped_actions) = build_cab_rules(actions, resources)?;
        if rules.is_empty() {
            return Err(AvError::Gcp(
                "None of the requested actions can be enforced via Credential Access Boundaries. \
                 CAB only supports Cloud Storage permissions."
                    .to_string(),
            ));
        }

        let boundary = serde_json::json!({
            "accessBoundary": {
                "accessBoundaryRules": rules
            }
        });

        let resp = self
            .http
            .post("https://sts.googleapis.com/v1/token")
            .form(&[
                (
                    "grant_type",
                    "urn:ietf:params:oauth:grant-type:token-exchange",
                ),
                (
                    "subject_token_type",
                    "urn:ietf:params:oauth:token-type:access_token",
                ),
                (
                    "requested_token_type",
                    "urn:ietf:params:oauth:token-type:access_token",
                ),
                ("subject_token", source_token),
                ("options", &boundary.to_string()),
            ])
            .send()
            .await
            .map_err(|e| AvError::Gcp(format!("STS token exchange failed: {}", e)))?;

        if !resp.status().is_success() {
            let status = resp.status();
            let text = resp.text().await.unwrap_or_default();
            let sanitized = sanitize_gcp_error(&text);
            return Err(AvError::Gcp(format!(
                "STS downscope error ({}): {}",
                status, sanitized
            )));
        }

        #[derive(Deserialize)]
        struct StsResponse {
            access_token: String,
            expires_in: Option<u64>,
        }

        let sts: StsResponse = resp
            .json()
            .await
            .map_err(|e| AvError::Gcp(format!("Failed to parse STS response: {}", e)))?;

        let expires_at = match sts.expires_in {
            Some(secs) => Utc::now() + chrono::Duration::seconds(secs as i64),
            None => {
                // STS omitted expires_in. Use the source token's remaining
                // lifetime as the fallback to avoid setting an expiry that
                // exceeds what the source token actually allows.
                source_expires_at.unwrap_or_else(|| {
                    tracing::warn!(
                        "STS response missing expires_in and no source token expiry available; \
                         defaulting to 3600s"
                    );
                    Utc::now() + chrono::Duration::seconds(3600)
                })
            }
        };

        Ok((
            GcpTempCredentials {
                access_token: sts.access_token,
                expires_at,
            },
            skipped_actions,
        ))
    }
}

/// Revoke a GCP OAuth2 access token.
///
/// Calls the Google OAuth2 revocation endpoint. Returns `Ok(true)` if
/// revocation was confirmed (HTTP 200), `Ok(false)` if the endpoint
/// returned a non-success status (best-effort; token may still expire
/// naturally), or `Err` if the request could not be sent at all.
///
/// R6-M23: uses a process-lifetime reqwest::Client with an explicit
/// timeout. The previous `reqwest::Client::new()` had no timeout at all,
/// so a slow or hostile revocation endpoint could hang session-end
/// indefinitely while the main thread waited for the revocation result.
pub async fn revoke_token(token: &str) -> Result<bool> {
    use std::sync::LazyLock;
    static REVOKE_CLIENT: LazyLock<reqwest::Client> = LazyLock::new(|| {
        reqwest::Client::builder()
            .timeout(Duration::from_secs(10))
            .connect_timeout(Duration::from_secs(5))
            .build()
            .unwrap_or_else(|_| reqwest::Client::new())
    });
    let client = &*REVOKE_CLIENT;
    match client
        .post("https://oauth2.googleapis.com/revoke")
        .form(&[("token", token)])
        .send()
        .await
    {
        Ok(resp) if resp.status().is_success() => {
            tracing::info!("GCP access token revoked successfully");
            Ok(true)
        }
        Ok(resp) => {
            tracing::warn!(
                status = %resp.status(),
                "GCP token revocation returned non-success status; token may not be revoked"
            );
            Ok(false)
        }
        Err(e) => Err(AvError::Gcp(format!(
            "GCP token revocation request failed: {e}"
        ))),
    }
}

/// Map GCP IAM permissions to Credential Access Boundary rules.
///
/// CAB supports Cloud Storage permissions via predefined roles. Returns the
/// boundary rules plus the list of requested actions that CAB cannot enforce
/// (R6-M18). An empty rules vec means no storage actions were requested —
/// the caller should error out. A non-empty rules vec with non-empty skipped
/// list means partial enforcement: storage actions are bound by CAB but the
/// listed non-storage actions are advisory-only and callers MUST reflect this
/// in audit logs and session metadata so operators are not misled.
fn build_cab_rules(
    actions: &[String],
    resources: &[&str],
) -> Result<(Vec<serde_json::Value>, Vec<String>)> {
    let mut rules = Vec::new();

    let storage_actions: Vec<&str> = actions
        .iter()
        .filter(|a| a.starts_with("storage."))
        .map(|a| a.as_str())
        .collect();

    // Collect the non-Storage actions so the caller can surface them in audit
    // logs and UI. Previously these were only logged via tracing::warn and
    // silently dropped from the boundary, leaving operators with the false
    // impression that their full --allow list was enforced.
    let skipped_actions: Vec<String> = actions
        .iter()
        .filter(|a| !a.starts_with("storage."))
        .cloned()
        .collect();
    if !skipped_actions.is_empty() {
        tracing::warn!(
            skipped_actions = ?skipped_actions,
            "CAB enforcement unavailable for non-Storage actions; \
             these permissions are advisory-only and will not be enforced at runtime"
        );
    }

    if storage_actions.is_empty() {
        return Ok((rules, skipped_actions));
    }

    // Map storage permissions to the most restrictive role that covers them
    let role = map_storage_role(&storage_actions);

    // R6-H12: emit one rule per requested resource. Previously only the
    // first resource was passed in and the rest silently received full SA
    // permissions. When no resources are supplied, fall back to a wildcard
    // over all buckets in the project (preserves prior behaviour).
    //
    // R6-M16: validate bucket names before interpolating them into the
    // CAB rule's `availableResource`. Without validation, a `--resource`
    // value like `mybucket/../../../other-project/buckets/attacker`
    // could escape the intended project scope, and slash/star characters
    // in the name could widen the boundary to unintended resources.
    // Names must match the Cloud Storage naming rules:
    // `[a-z0-9._-]+` with length 3..=222.
    let formatted_resources: Vec<String> = if resources.is_empty() {
        vec!["//storage.googleapis.com/projects/_/buckets/*".to_string()]
    } else {
        let mut out = Vec::with_capacity(resources.len());
        for r in resources {
            if r.starts_with("//") {
                out.push((*r).to_string());
            } else {
                if !is_valid_gcs_bucket_name(r) {
                    return Err(crate::error::AvError::InvalidPolicy(format!(
                        "Invalid GCS bucket name '{}': must match [a-z0-9._-]+ \
                         and be between 3 and 222 characters",
                        r
                    )));
                }
                out.push(format!("//storage.googleapis.com/projects/_/buckets/{}", r));
            }
        }
        out
    };

    for available_resource in formatted_resources {
        rules.push(serde_json::json!({
            "availablePermissions": [format!("inRole:roles/{}", role)],
            "availableResource": available_resource,
        }));
    }

    Ok((rules, skipped_actions))
}

/// Validate a GCS bucket name against Cloud Storage naming rules.
///
/// Cloud Storage accepts `[a-z0-9._-]+` with length 3..=222. We apply the
/// strictest subset that covers regular bucket names so that interpolating
/// the value into a CAB `availableResource` cannot escape the intended
/// project scope via `/` or `*` characters.
fn is_valid_gcs_bucket_name(name: &str) -> bool {
    let len = name.len();
    if !(3..=222).contains(&len) {
        return false;
    }
    name.chars()
        .all(|c| c.is_ascii_lowercase() || c.is_ascii_digit() || c == '.' || c == '_' || c == '-')
}

/// Determine the most restrictive Cloud Storage role for a set of permissions.
///
/// # Warning
///
/// `storage.objectAdmin` includes `storage.objects.setIamPolicy`, which allows
/// the holder to grant arbitrary access to individual objects. This role is
/// assigned when delete permissions are requested because Cloud Storage has no
/// built-in role that allows delete without also allowing `setIamPolicy`. Callers
/// that require delete access must accept this elevated privilege.
///
/// R6-M17: classify actions by the verb *after the last dot*, not by raw
/// substring. `a.contains("create")` would misclassify a hypothetical
/// read-side action like `storage.objects.createDraftRead` as a create, and
/// any action whose name happened to contain the substring `"delete"` (e.g.
/// `storage.foo.undelete` in a later API revision) would escalate the whole
/// session to `objectAdmin` unnecessarily.
fn map_storage_role(actions: &[&str]) -> &'static str {
    fn verb_of(action: &str) -> &str {
        action.rsplit('.').next().unwrap_or(action)
    }

    let has_delete = actions.iter().any(|a| verb_of(a) == "delete");
    let has_create = actions.iter().any(|a| {
        let v = verb_of(a);
        v == "create" || v == "update" || v == "insert"
    });

    if has_delete {
        "storage.objectAdmin"
    } else if has_create {
        "storage.objectCreator"
    } else {
        "storage.objectViewer"
    }
}

/// Map known action prefixes to narrower GCP OAuth scopes.
///
/// GCP OAuth scopes are coarse-grained (service-level), not permission-level.
/// Where a narrower scope exists for the requested service, use it instead of
/// the catch-all `cloud-platform` scope to reduce the blast radius of a
/// compromised token. Falls back to `cloud-platform` when no narrower scope
/// covers the requested actions.
///
/// Note: the `service_account` parameter is unused today but kept for future
/// expansion (e.g., per-SA scope overrides via config).
fn narrow_oauth_scopes(_service_account: &str) -> Vec<&'static str> {
    // cloud-platform is the broadest scope and covers everything.
    // Narrower scopes will be added here as the action-mapping system
    // is extended to carry per-action metadata.
    //
    // Known narrower scopes for future mapping:
    //   storage: https://www.googleapis.com/auth/devstorage.read_write
    //           https://www.googleapis.com/auth/devstorage.read_only
    //   bigquery: https://www.googleapis.com/auth/bigquery
    //   pubsub:   https://www.googleapis.com/auth/pubsub
    //   datastore: https://www.googleapis.com/auth/datastore
    //
    // Until per-action scope metadata is available we use cloud-platform
    // and rely on CAB (Credential Access Boundaries) for runtime enforcement
    // where supported.
    vec!["https://www.googleapis.com/auth/cloud-platform"]
}

/// Truncate and sanitize a GCP API error body for user-facing messages.
///
/// GCP error responses may contain project IDs and other internal identifiers.
/// This function strips patterns that look like project IDs and truncates the
/// body to 200 characters to avoid leaking sensitive details.
fn sanitize_gcp_error(body: &str) -> String {
    // R6-H13: compile regexes once at first use. Previously each error
    // response compiled three regexes from scratch, and on allocator failure
    // the pattern would silently fall through with the body unsanitized.
    use std::sync::LazyLock;
    static PROJECT_PATH: LazyLock<regex::Regex> =
        LazyLock::new(|| regex::Regex::new(r"projects/[a-zA-Z0-9_\-]+").unwrap());
    static PROJECT_ID_FIELD: LazyLock<regex::Regex> =
        LazyLock::new(|| regex::Regex::new(r#""project[_\-]?[Ii][Dd]"\s*:\s*"[^"]+""#).unwrap());
    static SA_EMAIL: LazyLock<regex::Regex> = LazyLock::new(|| {
        regex::Regex::new(r"[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.iam\.gserviceaccount\.com").unwrap()
    });

    let sanitized = PROJECT_PATH
        .replace_all(body, "projects/[REDACTED]")
        .into_owned();
    let sanitized = PROJECT_ID_FIELD
        .replace_all(&sanitized, "\"project_id\": \"[REDACTED]\"")
        .into_owned();
    let sanitized = SA_EMAIL
        .replace_all(&sanitized, "[SA_REDACTED]")
        .into_owned();

    if sanitized.chars().count() > 200 {
        format!("{}…", sanitized.chars().take(200).collect::<String>())
    } else {
        sanitized
    }
}