durable-actors 0.4.4

use std::sync::Arc;

use anyhow::{Result, ensure};
use aws_lc_rs::hmac;
use base64::{Engine, engine::general_purpose::URL_SAFE_NO_PAD};
use serde::{Deserialize, Serialize};

use crate::clock::Clock;

#[derive(Clone)]
pub struct ReplicaAccess {
    key: hmac::Key,
    secret: Arc<str>,
    clock: Arc<dyn Clock>,
}

#[derive(Clone, Debug, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct ReplicaGrant {
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub stream: Option<super::ReplicaStream>,
    pub operation: String,
    pub object: String,
    pub region: String,
    pub host_id: String,
    pub expires_at_ms: u64,
}

impl ReplicaAccess {
    pub fn new(secret: &str, clock: Arc<dyn Clock>) -> Self {
        Self {
            key: hmac::Key::new(hmac::HMAC_SHA256, secret.as_bytes()),
            secret: secret.into(),
            clock,
        }
    }

    pub fn secret(&self) -> &str {
        &self.secret
    }

    pub fn url(&self, origin: &str, grant: &ReplicaGrant) -> Result<String> {
        let payload = URL_SAFE_NO_PAD.encode(serde_json::to_vec(grant)?);
        let signature = URL_SAFE_NO_PAD.encode(hmac::sign(&self.key, payload.as_bytes()).as_ref());
        let url = reqwest::Url::parse(origin)?;
        ensure!(
            url.username().is_empty()
                && url.password().is_none()
                && url.query().is_none()
                && url.fragment().is_none()
                && url.path() == "/"
                && url.host_str().is_some(),
            "invalid replica origin"
        );
        let scheme = match url.scheme() {
            "http" => "grpc",
            "https" => "grpcs",
            _ => anyhow::bail!("replica origin must be HTTP or HTTPS"),
        };
        let origin = url.origin().ascii_serialization();
        let address = origin.replacen(url.scheme(), scheme, 1);
        Ok(format!("{address}?token={payload}.{signature}"))
    }

    pub(crate) fn verify(&self, token: &str, operation: &str) -> Result<ReplicaGrant> {
        self.verify_any(token, &[operation])
    }

    pub(crate) fn verify_any(&self, token: &str, operations: &[&str]) -> Result<ReplicaGrant> {
        ensure!(token.len() <= 16 * 1024, "replica capability is too large");
        let (payload, signature) = token
            .split_once('.')
            .ok_or_else(|| anyhow::anyhow!("invalid replica capability"))?;
        hmac::verify(
            &self.key,
            payload.as_bytes(),
            &URL_SAFE_NO_PAD.decode(signature)?,
        )
        .map_err(|_| anyhow::anyhow!("invalid replica signature"))?;
        let grant: ReplicaGrant = serde_json::from_slice(&URL_SAFE_NO_PAD.decode(payload)?)?;
        ensure!(
            operations.contains(&grant.operation.as_str()),
            "replica operation is not authorized"
        );
        ensure!(
            grant.expires_at_ms > self.clock.now_ms()?,
            "replica capability expired"
        );
        ensure!(
            grant.object.starts_with(crate::storage_paths::ROOT) && grant.object.len() <= 1024,
            "invalid replica object"
        );
        ensure!(
            grant
                .stream
                .as_ref()
                .is_none_or(|stream| stream.prefix == grant.object
                    && stream
                        .session
                        .starts_with(&format!("{}hosts/", crate::storage_paths::ROOT))
                    && stream.owner_epoch > 0),
            "replica stream does not match its capability"
        );
        crate::placement::validate_region(&grant.region)?;
        Ok(grant)
    }
}

#[cfg(test)]
#[path = "../../tests/unit/replication/access.rs"]
mod tests;