multistore 0.7.1

//! Authorization — checking if an identity can perform an operation.

use crate::error::ProxyError;
use crate::types::{Action, ResolvedIdentity, S3Operation};

/// Check if a key falls under an authorized prefix.
///
/// If the prefix already ends with `/`, a plain `starts_with` is sufficient.
/// Otherwise we require that the key either equals the prefix exactly or
/// that the character immediately after the prefix is `/`. This prevents
/// a prefix like `data` from matching `data-private/secret.txt`.
fn key_matches_prefix(key: &str, prefix: &str) -> bool {
    if prefix.ends_with('/') || prefix.is_empty() {
        return key.starts_with(prefix);
    }
    // Prefix does not end with '/' — require an exact match or a '/' boundary.
    // Done without allocating: this runs per key × scope × prefix on batch ops.
    key == prefix
        || (key.len() > prefix.len()
            && key.starts_with(prefix)
            && key.as_bytes()[prefix.len()] == b'/')
}

/// Check if a resolved identity is authorized to perform an operation.
pub fn authorize(
    identity: &ResolvedIdentity,
    operation: &S3Operation,
    bucket_config: &crate::types::BucketConfig,
) -> Result<(), ProxyError> {
    // Anonymous access check
    if matches!(identity, ResolvedIdentity::Anonymous) {
        if bucket_config.anonymous_access {
            // Anonymous users can only read
            let action = operation.action();
            if matches!(
                action,
                Action::GetObject | Action::HeadObject | Action::ListBucket
            ) {
                return Ok(());
            }
        }
        return Err(ProxyError::AccessDenied);
    }

    let scopes = match identity {
        ResolvedIdentity::Anonymous => unreachable!(),
        ResolvedIdentity::Authenticated(id) => &id.allowed_scopes,
    };

    let action = operation.action();
    let bucket = operation.bucket().unwrap_or_default().to_string();
    let key = match operation {
        S3Operation::ListBucket { raw_query, .. } => {
            // Extract prefix from raw query for authorization checks
            raw_query
                .as_deref()
                .and_then(|q| {
                    url::form_urlencoded::parse(q.as_bytes())
                        .find(|(k, _)| k == "prefix")
                        .map(|(_, v)| v.to_string())
                })
                .unwrap_or_default()
        }
        _ => operation.key().to_string(),
    };

    // Batch delete carries no key here — the keys live in the request body. This
    // is only a coarse check that *some* scope grants DeleteObject on the bucket;
    // each key is authorized individually (against its prefix) once the body is
    // parsed. See [`key_authorized`].
    let ignore_prefix = matches!(operation, S3Operation::DeleteObjects { .. });

    // Check if any scope grants access
    let authorized = scopes.iter().any(|scope| {
        if scope.bucket != bucket {
            return false;
        }
        if !scope.actions.contains(&action) {
            return false;
        }
        // Check prefix restrictions
        if ignore_prefix || scope.prefixes.is_empty() {
            return true; // Full bucket access (or deferred per-key check)
        }
        scope
            .prefixes
            .iter()
            .any(|prefix| key_matches_prefix(&key, prefix))
    });

    if authorized {
        Ok(())
    } else {
        tracing::warn!(
            action = ?action,
            bucket = %bucket,
            key = %key,
            scopes = ?scopes,
            "authorization denied — no scope grants access"
        );
        Err(ProxyError::AccessDenied)
    }
}

/// Check whether `identity` may perform `action` on a single `key` in `bucket`.
///
/// Used for per-key authorization of batch operations such as
/// [`DeleteObjects`](crate::types::S3Operation::DeleteObjects), where the coarse
/// [`authorize`] check only verified that *some* scope grants the action on the
/// bucket. Anonymous identities are never authorized here — this is only used
/// for write actions, which anonymous callers can never perform.
pub fn key_authorized(
    identity: &ResolvedIdentity,
    bucket: &str,
    action: Action,
    key: &str,
) -> bool {
    let scopes = match identity {
        ResolvedIdentity::Anonymous => return false,
        ResolvedIdentity::Authenticated(id) => &id.allowed_scopes,
    };
    scopes.iter().any(|scope| {
        scope.bucket == bucket
            && scope.actions.contains(&action)
            && (scope.prefixes.is_empty()
                || scope.prefixes.iter().any(|p| key_matches_prefix(key, p)))
    })
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn prefix_with_slash_matches_children() {
        assert!(key_matches_prefix("data/file.txt", "data/"));
        assert!(key_matches_prefix("data/sub/file.txt", "data/"));
    }

    #[test]
    fn prefix_without_slash_enforces_boundary() {
        assert!(key_matches_prefix("data/file.txt", "data"));
        assert!(key_matches_prefix("data", "data"));
        assert!(!key_matches_prefix("data-private/secret.txt", "data"));
        assert!(!key_matches_prefix("database/dump.sql", "data"));
    }

    #[test]
    fn empty_prefix_matches_everything() {
        assert!(key_matches_prefix("anything/at/all.txt", ""));
        assert!(key_matches_prefix("", ""));
    }

    #[test]
    fn prefix_no_match() {
        assert!(!key_matches_prefix("other/file.txt", "data/"));
        assert!(!key_matches_prefix("other/file.txt", "data"));
    }

    use crate::types::{AccessScope, AuthenticatedIdentity, BucketConfig};

    fn identity_with(scope: AccessScope) -> ResolvedIdentity {
        ResolvedIdentity::Authenticated(AuthenticatedIdentity {
            principal_name: "tester".into(),
            allowed_scopes: vec![scope],
        })
    }

    fn bucket(name: &str, anonymous: bool) -> BucketConfig {
        BucketConfig {
            name: name.into(),
            backend_type: "s3".into(),
            backend_prefix: None,
            anonymous_access: anonymous,
            allowed_roles: vec![],
            backend_options: Default::default(),
        }
    }

    #[test]
    fn key_authorized_enforces_prefix_per_key() {
        let id = identity_with(AccessScope {
            bucket: "b".into(),
            prefixes: vec!["data/".into()],
            actions: vec![Action::DeleteObject],
        });
        assert!(key_authorized(&id, "b", Action::DeleteObject, "data/x.txt"));
        assert!(!key_authorized(
            &id,
            "b",
            Action::DeleteObject,
            "other/x.txt"
        ));
        // Wrong bucket / wrong action are denied.
        assert!(!key_authorized(
            &id,
            "other",
            Action::DeleteObject,
            "data/x.txt"
        ));
        assert!(!key_authorized(&id, "b", Action::PutObject, "data/x.txt"));
    }

    #[test]
    fn key_authorized_denies_anonymous() {
        assert!(!key_authorized(
            &ResolvedIdentity::Anonymous,
            "b",
            Action::DeleteObject,
            "anything"
        ));
    }

    #[test]
    fn delete_objects_coarse_authz_ignores_prefix() {
        // A prefix-scoped caller passes the coarse batch-delete check even though
        // the operation carries no key; per-key enforcement happens later.
        let id = identity_with(AccessScope {
            bucket: "b".into(),
            prefixes: vec!["data/".into()],
            actions: vec![Action::DeleteObject],
        });
        let op = S3Operation::DeleteObjects { bucket: "b".into() };
        assert!(authorize(&id, &op, &bucket("b", false)).is_ok());
    }

    #[test]
    fn delete_objects_denied_without_delete_action() {
        let id = identity_with(AccessScope {
            bucket: "b".into(),
            prefixes: vec![],
            actions: vec![Action::GetObject],
        });
        let op = S3Operation::DeleteObjects { bucket: "b".into() };
        assert!(authorize(&id, &op, &bucket("b", false)).is_err());
    }

    #[test]
    fn delete_objects_denied_for_anonymous() {
        let op = S3Operation::DeleteObjects { bucket: "b".into() };
        // Even on an anonymous-readable bucket, batch delete is a write.
        assert!(authorize(&ResolvedIdentity::Anonymous, &op, &bucket("b", true)).is_err());
    }

    fn copy(dest_key: &str) -> S3Operation {
        S3Operation::CopyObject {
            bucket: "b".into(),
            key: dest_key.into(),
            src_bucket: "b".into(),
            src_key: "any/source.txt".into(),
            src_version: None,
        }
    }

    /// A copy is a write of its *destination* key, so a prefix-scoped caller may
    /// only copy into the prefixes they hold. Were the operation's key to read
    /// as anything else (the source key, or empty), a scoped caller could write
    /// anywhere in the bucket.
    #[test]
    fn copy_object_enforces_prefix_on_the_destination_key() {
        let id = identity_with(AccessScope {
            bucket: "b".into(),
            prefixes: vec!["public/".into()],
            actions: vec![Action::PutObject],
        });
        assert!(authorize(&id, &copy("public/x.txt"), &bucket("b", false)).is_ok());
        assert!(authorize(&id, &copy("private/x.txt"), &bucket("b", false)).is_err());
    }

    /// Read access is not copy access: `CopyObject` maps to `PutObject`, so a
    /// caller holding only `GetObject` cannot copy even within their prefix.
    #[test]
    fn copy_object_requires_the_put_action() {
        let id = identity_with(AccessScope {
            bucket: "b".into(),
            prefixes: vec![],
            actions: vec![Action::GetObject],
        });
        assert!(authorize(&id, &copy("anywhere.txt"), &bucket("b", false)).is_err());
    }

    /// A copy writes, so an anonymous caller is denied even on a bucket that
    /// allows anonymous reads.
    #[test]
    fn copy_object_denied_for_anonymous() {
        assert!(authorize(
            &ResolvedIdentity::Anonymous,
            &copy("x.txt"),
            &bucket("b", true)
        )
        .is_err());
    }

    /// The hole this exists to close: on the bundled policy, a caller granted
    /// `get_object` over a prefix must not be able to read — or copy out — an
    /// older version of an object in it. A version-scoped read is its own
    /// action, so the `get_object` grant simply does not match.
    #[test]
    fn get_object_grant_does_not_permit_a_versioned_read() {
        let id = identity_with(AccessScope {
            bucket: "b".into(),
            prefixes: vec!["public/".into()],
            actions: vec![Action::GetObject, Action::PutObject],
        });
        let versioned = S3Operation::GetObject {
            bucket: "b".into(),
            key: "public/src.txt".into(),
            version: Some("v1".into()),
        };
        assert!(
            authorize(&id, &versioned, &bucket("b", false)).is_err(),
            "get_object must not imply reading previous versions"
        );
    }

    /// ...and a caller who *is* granted the version action can, within their
    /// prefix. Denying by default must stay expressible, not absolute.
    #[test]
    fn get_object_version_grant_permits_a_versioned_read_in_prefix() {
        let id = identity_with(AccessScope {
            bucket: "b".into(),
            prefixes: vec!["public/".into()],
            actions: vec![Action::GetObjectVersion],
        });
        let versioned = |key: &str| S3Operation::GetObject {
            bucket: "b".into(),
            key: key.into(),
            version: Some("v1".into()),
        };
        assert!(authorize(&id, &versioned("public/src.txt"), &bucket("b", false)).is_ok());
        // The prefix still binds.
        assert!(authorize(&id, &versioned("private/src.txt"), &bucket("b", false)).is_err());
    }

    /// An anonymous caller never reads a previous version, even on a bucket
    /// that allows anonymous reads: the anonymous grant covers current-object
    /// reads only.
    #[test]
    fn versioned_read_denied_for_anonymous() {
        let versioned = S3Operation::GetObject {
            bucket: "b".into(),
            key: "x.txt".into(),
            version: Some("v1".into()),
        };
        assert!(authorize(&ResolvedIdentity::Anonymous, &versioned, &bucket("b", true)).is_err());
    }

    /// The other half of a copy's authorization: the gateway resolves the copy
    /// source through a synthetic `GetObject`, so the source key is prefix-
    /// checked as a read. A caller scoped to `public/` cannot name a source
    /// outside it.
    #[test]
    fn copy_source_read_enforces_prefix_via_get_object() {
        let id = identity_with(AccessScope {
            bucket: "b".into(),
            prefixes: vec!["public/".into()],
            actions: vec![Action::GetObject, Action::PutObject],
        });
        let read = |key: &str| S3Operation::GetObject {
            bucket: "b".into(),
            key: key.into(),
            version: None,
        };
        assert!(authorize(&id, &read("public/src.txt"), &bucket("b", false)).is_ok());
        assert!(authorize(&id, &read("private/src.txt"), &bucket("b", false)).is_err());
    }
}