use crate::api::tables::{
DirectRead as IssuedDirectRead, StorageAccess, StorageBinding as IssuedBinding, StorageMode,
VendedCredential,
};
use crate::scandal::{
DirectRead, GcsSession, S3Session, ScandalError, StorageBinding, StorageCredential,
};
impl StorageBinding {
pub fn from_storage(root: String, binding: IssuedBinding) -> Result<Self, ScandalError> {
let expires_at = Some(binding.expires_at);
match binding.access {
StorageAccess::Signed(access) => {
let direct_read = match access.direct_read {
Some(direct) if binding.mode == StorageMode::ReadOnly => {
Some(direct_read(direct, &root)?)
}
_ => None,
};
Ok(Self {
root_alias: (root != binding.root_uri).then_some(binding.root_uri),
root_uri: root,
credential: Some(StorageCredential(access.token.0)),
endpoint: (!access.endpoint.is_empty()).then_some(access.endpoint),
expires_at,
direct_read,
})
}
StorageAccess::Vended(access) => {
let credential = session(access.credential, &root)?;
let root_alias = (root != binding.root_uri).then_some(root);
Ok(Self {
root_uri: binding.root_uri,
root_alias,
credential: Some(credential),
endpoint: None,
expires_at,
direct_read: None,
})
}
StorageAccess::Unknown { .. } => Err(unsupported(&root)),
}
}
pub fn for_resource(binding: IssuedBinding) -> Result<Self, ScandalError> {
Self::from_storage(binding.root_uri.clone(), binding)
}
}
fn direct_read(direct: IssuedDirectRead, root: &str) -> Result<DirectRead, ScandalError> {
Ok(DirectRead {
root_uri: direct.uri,
credential: session(direct.credential, root)?,
})
}
fn session(credential: VendedCredential, root: &str) -> Result<StorageCredential, ScandalError> {
let json = match credential {
VendedCredential::AwsSession(credential) => serde_json::to_string(&S3Session {
access_key_id: credential.access_key_id,
secret_access_key: credential.secret_access_key.0,
session_token: credential.session_token.0,
region: credential.region,
endpoint: credential.endpoint,
expires_at_micros: credential.expires_at_micros.0,
}),
VendedCredential::GcsAccessToken(credential) => serde_json::to_string(&GcsSession {
access_token: credential.access_token.0,
expires_at_micros: credential.expires_at_micros.0,
}),
_ => return Err(unsupported(root)),
};
json.map(StorageCredential)
.map_err(|error| ScandalError::Internal {
message: error.to_string(),
})
}
fn unsupported(root: &str) -> ScandalError {
ScandalError::InvalidArgument {
message: format!("storage at {root} is neither signed by SPFS nor in object storage"),
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::api::tables::{
GcsAccessToken, HuggingFaceToken, SecretToken, SignedAccess, VendedAccess,
};
fn vended_binding(credential: VendedCredential) -> IssuedBinding {
IssuedBinding {
root_uri: "gs://acme/lake/tables/t1/".to_owned(),
mode: StorageMode::ReadOnly,
expires_at: 0.into(),
access: StorageAccess::Vended(VendedAccess { credential }),
}
}
#[test]
fn a_vended_binding_carries_its_session_to_scandal() {
let root = "gs://acme/lake/tables/t1/".to_owned();
let gcs = || {
vended_binding(VendedCredential::GcsAccessToken(GcsAccessToken {
access_token: SecretToken("ya29.vended".into()),
expires_at_micros: 7.into(),
}))
};
let binding = StorageBinding::from_storage(root.clone(), gcs()).unwrap();
assert_eq!(binding.root_uri, root);
assert_eq!(binding.root_alias, None);
let session: GcsSession = serde_json::from_str(binding.credential().unwrap()).unwrap();
assert_eq!(session.access_token, "ya29.vended");
assert_eq!(session.expires_at_micros, 7);
let aliased =
StorageBinding::from_storage("spfsu://4b44-alias/".to_owned(), gcs()).unwrap();
assert_eq!(aliased.root_uri, root);
assert_eq!(aliased.root_alias.as_deref(), Some("spfsu://4b44-alias/"));
let refused = StorageBinding::from_storage(
root,
vended_binding(VendedCredential::HuggingFaceToken(HuggingFaceToken {
token: SecretToken("hf".into()),
expires_at_micros: 7.into(),
endpoint: "https://huggingface.co".to_owned(),
})),
);
assert!(refused.is_err());
}
fn signed_binding(mode: StorageMode, endpoint: &str) -> IssuedBinding {
IssuedBinding {
root_uri: "spfsu://4b44-alias/".to_owned(),
mode,
expires_at: 1_900_000_000.into(),
access: StorageAccess::Signed(SignedAccess {
endpoint: endpoint.to_owned(),
token: SecretToken("capability".into()),
direct_read: Some(IssuedDirectRead {
uri: "gs://acme/lake/tables/t1/".to_owned(),
credential: VendedCredential::GcsAccessToken(GcsAccessToken {
access_token: SecretToken("ya29.vended".into()),
expires_at_micros: 7.into(),
}),
}),
}),
}
}
#[test]
fn a_signed_binding_carries_its_terms_to_scandal() {
let binding = StorageBinding::for_resource(signed_binding(
StorageMode::ReadOnly,
"https://spfs.example/",
))
.unwrap();
assert_eq!(binding.endpoint.as_deref(), Some("https://spfs.example/"));
assert_eq!(binding.expires_at.map(u64::from), Some(1_900_000_000));
let direct = binding.direct_read.unwrap();
assert_eq!(direct.root_uri, "gs://acme/lake/tables/t1/");
let session: GcsSession = serde_json::from_str(&direct.credential.0).unwrap();
assert_eq!(session.access_token, "ya29.vended");
let writable =
StorageBinding::for_resource(signed_binding(StorageMode::ReadWrite, "")).unwrap();
assert_eq!(writable.direct_read, None);
assert_eq!(writable.endpoint, None);
}
}