spiral-sdk 0.1.7

//! URI resolution owned by an authenticated SDK host.
use async_trait::async_trait;
use reqwest::{Client as ReqwestClient, StatusCode};
use serde::Deserialize;
use url::Url;
use uuid::Uuid;

use crate::Config;
use crate::api::tables::{
    AwsSession, GcsAccessToken, SecretToken, SignedAccess, StorageAccess,
    StorageBinding as IssuedBinding, StorageMode, VendedAccess, VendedCredential,
};
use crate::auth::TokenSource;
use crate::scandal::error::ScandalError;
use crate::scandal::{ResolvedUri, StorageBinding, UriResolver};

/// Resolve Spiral resource addresses using the SDK host's shared credentials and HTTP pool.
/// Provider URIs are not implicitly authorized; their policy requires a separate host resolver.
#[derive(Clone)]
pub struct ResourceUriResolver {
    api_url: Url,
    auth: TokenSource,
    http: ReqwestClient,
}

impl ResourceUriResolver {
    /// Use the same endpoint, token provider and HTTP pool as `config`'s resource clients.
    pub fn from_config(config: &Config) -> Self {
        Self::new(
            config.api_url.clone(),
            config.auth.clone(),
            config.http.clone(),
        )
    }

    /// Resolve as the identity `auth` provides, through the stack at `api_url`. `http` must
    /// not follow redirects, since requests carry the identity's bearer token.
    pub fn new(api_url: Url, auth: TokenSource, http: ReqwestClient) -> Self {
        Self {
            api_url,
            auth,
            http,
        }
    }
}

#[async_trait]
impl UriResolver for ResourceUriResolver {
    async fn resolve(&self, uri: &Url) -> Result<ResolvedUri, ScandalError> {
        if uri.scheme() != "spiral"
            || uri.query().is_some()
            || uri.fragment().is_some()
            || !uri.username().is_empty()
            || uri.password().is_some()
        {
            return Err(ScandalError::InvalidArgument {
                message: "this host resolves only spiral:// resource addresses".into(),
            });
        }
        let resource = uri
            .host_str()
            .and_then(|host| Uuid::parse_str(host).ok())
            .ok_or_else(|| ScandalError::InvalidArgument {
                message: "Spiral address requires a resource UUID".into(),
            })?;
        let token = self.auth.token().await.map_err(internal)?.ok_or_else(|| {
            ScandalError::Unauthenticated {
                message: "URI resolution requires the host's credentials".into(),
            }
        })?;
        let response = self
            .http
            .post(format!(
                "{}/v1/resources/{resource}/storage-binding",
                self.api_url.as_str().trim_end_matches('/')
            ))
            .bearer_auth(token.expose_secret())
            .json(&serde_json::json!({ "mode": "readOnly" }))
            .send()
            .await
            .map_err(unavailable)?;
        match response.status() {
            status if status.is_success() => {}
            StatusCode::BAD_REQUEST => {
                let refusal: BindingRefusal = bounded_json(response).await?;
                return Err(ScandalError::InvalidArgument {
                    message: refusal.message,
                });
            }
            StatusCode::UNAUTHORIZED => {
                return Err(ScandalError::Unauthenticated {
                    message: "storage refused the host's credential".into(),
                });
            }
            StatusCode::FORBIDDEN => {
                return Err(ScandalError::PermissionDenied {
                    operation: "read".into(),
                    resource: resource.to_string(),
                });
            }
            StatusCode::NOT_FOUND => {
                return Err(ScandalError::NotFound {
                    resource: resource.to_string(),
                });
            }
            status => return Err(unavailable(format!("storage returned {status}"))),
        }
        let binding: ResourceBinding = bounded_json(response).await?;
        if binding.mode != "readOnly" {
            return Err(internal("storage returned a writable URI binding"));
        }
        Ok(ResolvedUri {
            root_uri: uri.join("/").map_err(internal)?,
            storage: StorageBinding::for_resource(binding.into_issued())?,
        })
    }
}

// The resource binding route uses flat camelCase wire objects.
#[derive(Deserialize)]
#[serde(rename_all = "camelCase")]
struct ResourceBinding {
    root_uri: String,
    mode: String,
    expires_at: u64,
    access: ResourceAccess,
}

#[derive(Deserialize)]
#[serde(tag = "type", rename_all = "camelCase")]
enum ResourceAccess {
    Signed { token: SecretToken },
    Vended { credential: ResourceCredential },
}

#[derive(Deserialize)]
#[serde(tag = "type", rename_all = "camelCase")]
enum ResourceCredential {
    AwsSession(AwsSession),
    GcsAccessToken(GcsAccessToken),
}

impl ResourceBinding {
    fn into_issued(self) -> IssuedBinding {
        let access = match self.access {
            ResourceAccess::Signed { token } => StorageAccess::Signed(SignedAccess {
                endpoint: String::new(),
                token,
                direct_read: None,
            }),
            ResourceAccess::Vended { credential } => {
                let credential = match credential {
                    ResourceCredential::AwsSession(session) => {
                        VendedCredential::AwsSession(session)
                    }
                    ResourceCredential::GcsAccessToken(token) => {
                        VendedCredential::GcsAccessToken(token)
                    }
                };
                StorageAccess::Vended(VendedAccess { credential })
            }
        };
        IssuedBinding {
            root_uri: self.root_uri,
            mode: StorageMode::ReadOnly,
            expires_at: self.expires_at.into(),
            access,
        }
    }
}

#[derive(Deserialize)]
struct BindingRefusal {
    message: String,
}

async fn bounded_json<T: for<'de> Deserialize<'de>>(
    response: reqwest::Response,
) -> Result<T, ScandalError> {
    let mut response = response;
    let mut bytes = Vec::new();
    while let Some(chunk) = response.chunk().await.map_err(unavailable)? {
        if bytes.len() + chunk.len() > 1024 * 1024 {
            return Err(internal("oversized host credential response"));
        }
        bytes.extend_from_slice(&chunk);
    }
    serde_json::from_slice(&bytes).map_err(internal)
}

fn internal(error: impl ToString) -> ScandalError {
    ScandalError::Internal {
        message: error.to_string(),
    }
}

fn unavailable(error: impl ToString) -> ScandalError {
    ScandalError::Unavailable {
        message: error.to_string(),
    }
}