spiral/

storage_binding.rs

//! Scandal bindings from the storage bindings that Spiral's storage service issues.

use crate::api::tables::{
    DirectRead as IssuedDirectRead, StorageAccess, StorageBinding as IssuedBinding, StorageMode,
    VendedCredential,
};
use crate::scandal::{
    DirectRead, GcsSession, S3Session, ScandalError, StorageBinding, StorageCredential,
};

impl StorageBinding {
    /// The scandal binding of `root` that storage's `binding` authorizes, with its deadline: an
    /// SPFS capability, signed at the binding's endpoint and read directly at the provider when
    /// a read-only binding carries a direct read, or a provider session vended for the binding's
    /// provider root. When `root`, the root that a Table's object references name, differs from
    /// the binding's, it becomes the alias.
    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)),
                    // Storage access of older stacks names no endpoint.
                    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)),
        }
    }

    /// The scandal binding of a resource's whole storage, which `spiral://` addresses name
    /// relative to its 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)?,
    })
}

/// The provider session `credential` grants, as scandal reads it.
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 }),
        }
    }

    /// A vended binding reaches scandal at its provider root with its session as the
    /// credential, aliased by the table's recorded root when that differs, and a credential that
    /// opens no object store is refused.
    #[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(),
                    }),
                }),
            }),
        }
    }

    /// A signed binding reaches scandal with the endpoint that signs it, its deadline and,
    /// read-only, its direct read; older stacks name no endpoint.
    #[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);
    }
}