spiral-sdk 0.1.11

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

use crate::Config;
use crate::api::tables::{
    AwsSession, DirectRead, GcsAccessToken, SecretToken, SignedAccess, StorageAccess,
    StorageBinding as IssuedBinding, StorageMode, VendedAccess, VendedCredential,
};
use crate::auth::TokenSource;
use crate::scandal::error::ScandalError;
use crate::scandal::{ResolvedUri, ResourceAccess, 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: impl Into<ReqwestClient>) -> Self {
        let http = http.into();
        Self {
            api_url,
            auth,
            http,
        }
    }
}

#[async_trait]
impl UriResolver for ResourceUriResolver {
    async fn resolve(&self, uri: &Url) -> Result<ResolvedUri, ScandalError> {
        self.resolve_for(uri, ResourceAccess::Read).await
    }

    async fn resolve_for(
        &self,
        uri: &Url,
        access: ResourceAccess,
    ) -> Result<ResolvedUri, ScandalError> {
        if uri.scheme() != "spiral"
            || uri.query().is_some()
            || uri.fragment().is_some()
            || !uri.username().is_empty()
            || uri.password().is_some()
            || uri.port().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(&match access {
                ResourceAccess::Read => serde_json::json!({"mode":"readOnly"}),
                ResourceAccess::Attempt { attempt_id } => serde_json::json!({"mode":"readWrite", "write":{"type":"attempt","attemptId":attempt_id}}),
                ResourceAccess::Publication => serde_json::json!({"mode":"readWrite", "write":{"type":"publication"}}),
            })
            .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 (access == ResourceAccess::Read && binding.mode != "readOnly")
            || (access != ResourceAccess::Read && binding.mode != "readWrite")
        {
            return Err(internal("storage returned the wrong resource access mode"));
        }
        Ok(ResolvedUri {
            root_uri: uri.join("/").map_err(internal)?,
            storage: binding.into_storage()?,
        })
    }
}

// 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: ResourceStorageAccess,
}

#[derive(Deserialize)]
#[serde(tag = "type", rename_all = "camelCase")]
enum ResourceStorageAccess {
    #[serde(rename_all = "camelCase")]
    Signed {
        endpoint: String,
        token: SecretToken,
        #[serde(default)]
        direct_read: Option<ResourceDirectRead>,
    },
    Vended {
        credential: ResourceCredential,
    },
    Unauthenticated,
}

#[derive(Deserialize)]
struct ResourceDirectRead {
    uri: String,
    credential: ResourceCredential,
}

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

impl From<ResourceCredential> for VendedCredential {
    fn from(credential: ResourceCredential) -> Self {
        match credential {
            ResourceCredential::AwsSession(session) => Self::AwsSession(session),
            ResourceCredential::GcsAccessToken(token) => Self::GcsAccessToken(token),
        }
    }
}

impl ResourceBinding {
    fn into_storage(self) -> Result<StorageBinding, ScandalError> {
        let mode = if self.mode == "readOnly" {
            StorageMode::ReadOnly
        } else {
            StorageMode::ReadWrite
        };
        let access = match self.access {
            ResourceStorageAccess::Unauthenticated => {
                return Ok(StorageBinding::new(self.root_uri));
            }
            ResourceStorageAccess::Signed {
                endpoint,
                token,
                direct_read,
            } => StorageAccess::Signed(SignedAccess {
                endpoint,
                token,
                direct_read: direct_read.map(|direct| DirectRead {
                    uri: direct.uri,
                    credential: direct.credential.into(),
                }),
            }),
            ResourceStorageAccess::Vended { credential } => StorageAccess::Vended(VendedAccess {
                credential: credential.into(),
            }),
        };
        StorageBinding::for_resource(IssuedBinding {
            root_uri: self.root_uri,
            mode,
            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(),
    }
}