//! Role-based authorization for the gRPC surface
//! (architecture/roles-auth.md).
//!
//! Enforcement lives in exactly one place: [`AuthLayer`], a tower layer
//! in front of the tonic service. It authenticates the HTTP basic-auth
//! header against the configured role hashes and checks the resulting
//! [`Role`] against the method's [`minimum_role`] — *before* any
//! handler runs, default-deny for methods it does not know. Handlers
//! never see unauthorized requests and did not change for this feature.
//!
//! Credentials never appear in logs or error messages.

use std::collections::HashMap;
use std::sync::{Arc, RwLock};
use std::task::{Context, Poll};

use argon2::{Argon2, PasswordHash, PasswordVerifier};
use base64::Engine;
use futures::future::{ready, Either, Ready};
use tonic::Status;
use tracing::warn;

use crate::settings::AuthSettings;

/// The three roles, ordered by privilege: every role includes the
/// rights of the roles below it.
#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)]
pub enum Role {
    /// May read and append tracks to the queue (plus create search
    /// terms — the mechanism of finding something to append).
    QueueAppender,
    /// Anything on the queue and playback, but no library writes.
    QueueOwner,
    /// The normal user: everything.
    Owner,
}

impl Role {
    /// The basic-auth user name selecting this role.
    fn from_user(user: &str) -> Option<Role> {
        match user {
            "owner" => Some(Role::Owner),
            "queue-owner" => Some(Role::QueueOwner),
            "queue-appender" => Some(Role::QueueAppender),
            _ => None,
        }
    }

    fn name(self) -> &'static str {
        match self {
            Role::Owner => "owner",
            Role::QueueOwner => "queue-owner",
            Role::QueueAppender => "queue-appender",
        }
    }
}

/// The gRPC path prefix of our service's methods.
const SERVICE_PREFIX: &str = "/crabidy.v1.CrabidyService/";

/// Minimum role required for a gRPC request path (the rights matrix,
/// architecture/roles-auth.md). Unknown methods — including anything
/// outside our service — require [`Role::Owner`]: fail-closed, a
/// future RPC starts locked until it is mapped here (a test pins the
/// full method list, so forgetting fails the suite).
pub fn minimum_role(grpc_path: &str) -> Role {
    let Some(method) = grpc_path.strip_prefix(SERVICE_PREFIX) else {
        return Role::Owner;
    };
    match method {
        // Reads, the one appender queue verb, and search-term creation.
        "Init" | "GetLibraryNode" | "GetUpdateStream" | "Append" | "CreateLibraryNode" => {
            Role::QueueAppender
        }
        // Every other queue and playback verb.
        "Queue" | "Replace" | "Remove" | "Insert" | "ClearQueue" | "SetCurrent"
        | "ToggleShuffle" | "ToggleRepeat" | "TogglePlay" | "Stop" | "ChangeVolume"
        | "ToggleMute" | "Next" | "Prev" | "RestartTrack" => Role::QueueOwner,
        // Library writes (CaptureLibraryNode, SaveQueue,
        // RenameLibraryNode, DeleteLibraryNode) and anything unmapped.
        _ => Role::Owner,
    }
}

/// Hashes a password into the PHC string `crabidy-server.toml` expects
/// (argon2id, default parameters, fresh random salt). Backs the
/// `crabidy-server hash-password` helper.
pub fn hash_password(password: &str) -> Result<String, String> {
    use argon2::password_hash::{rand_core::OsRng, SaltString};
    use argon2::PasswordHasher;
    Argon2::default()
        .hash_password(password.as_bytes(), &SaltString::generate(&mut OsRng))
        // The error is parameter trouble, never the password itself.
        .map(|hash| hash.to_string())
        .map_err(|err| format!("cannot hash password: {err}"))
}

/// Verifies basic-auth credentials against the configured role hashes.
///
/// Argon2 verification is deliberately slow, so *successful* header
/// values are cached (value → role); the cache is fed only by
/// successes, bounding it by the number of valid credentials. Failures
/// re-verify every time, which doubles as throttling.
pub struct Authenticator {
    /// `(role, PHC hash)` pairs from the config; empty = auth off.
    hashes: Vec<(Role, String)>,
    /// The role granted to requests that carry no credentials: the most
    /// privileged role left *unguarded* (no hash in the config), or
    /// `None` when every role is guarded and anonymous access is denied.
    /// Guarding runs from the top down (`AuthSettings::validate`), so
    /// this is simply the first role, owner → queue-owner →
    /// queue-appender, whose hash is absent.
    unauthenticated: Option<Role>,
    verified: RwLock<HashMap<String, Role>>,
}

impl Authenticator {
    pub fn new(settings: &AuthSettings) -> Self {
        let mut hashes = Vec::new();
        for (role, hash) in [
            (Role::Owner, &settings.owner),
            (Role::QueueOwner, &settings.queue_owner),
            (Role::QueueAppender, &settings.queue_appender),
        ] {
            if let Some(hash) = hash {
                // Reject unusable hashes at startup, when the operator
                // is looking — not at the first login attempt.
                if let Err(err) = PasswordHash::new(hash) {
                    warn!(
                        role = role.name(),
                        "unusable password hash in config: {err}"
                    );
                } else {
                    hashes.push((role, hash.clone()));
                }
            }
        }
        // Anonymous callers inherit the highest role that is *not*
        // guarded. A hash that is present but unusable still counts as
        // guarded here (fail-closed): that level is neither reachable by
        // login nor handed to anonymous callers.
        let unauthenticated = if settings.owner.is_none() {
            Some(Role::Owner)
        } else if settings.queue_owner.is_none() {
            Some(Role::QueueOwner)
        } else if settings.queue_appender.is_none() {
            Some(Role::QueueAppender)
        } else {
            None
        };
        Self {
            hashes,
            unauthenticated,
            verified: RwLock::new(HashMap::new()),
        }
    }

    /// Whether any credential is configured (the auth on/off switch).
    pub fn enabled(&self) -> bool {
        !self.hashes.is_empty()
    }

    /// The role an anonymous (headerless) request receives — the most
    /// privileged unguarded role, or `None` when every role is guarded.
    pub fn unauthenticated_role(&self) -> Option<Role> {
        self.unauthenticated
    }

    /// Resolves the request's `authorization` header value to a role.
    ///
    /// A request with **no** `authorization` header is anonymous and
    /// receives [`Authenticator::unauthenticated_role`] — the most
    /// privileged unguarded role — or is denied when every role is
    /// guarded. A request that **does** carry a header is attempting to
    /// authenticate; every way that can fail — wrong scheme, broken
    /// base64, unknown user, wrong password — answers the same
    /// `UNAUTHENTICATED` so callers cannot probe which part was wrong.
    /// It never silently falls back to the anonymous role. Never panics
    /// on input.
    pub fn authenticate(&self, header: Option<&str>) -> Result<Role, Status> {
        let denied = || Status::unauthenticated("credentials required");
        let Some(header) = header else {
            return self.unauthenticated.ok_or_else(denied);
        };
        if let Some(role) = self
            .verified
            .read()
            .ok()
            .and_then(|cache| cache.get(header).copied())
        {
            return Ok(role);
        }
        let encoded = header
            .strip_prefix("Basic ")
            .or_else(|| header.strip_prefix("basic "))
            .ok_or_else(denied)?;
        let decoded = base64::engine::general_purpose::STANDARD
            .decode(encoded.trim())
            .map_err(|_| denied())?;
        let decoded = String::from_utf8(decoded).map_err(|_| denied())?;
        let (user, password) = decoded.split_once(':').ok_or_else(denied)?;
        let role = Role::from_user(user).ok_or_else(denied)?;
        let hash = self
            .hashes
            .iter()
            .find(|(r, _)| *r == role)
            .map(|(_, h)| h)
            .ok_or_else(denied)?;
        // Validated in `new`; a parse failure here is unreachable but
        // must still deny, not panic.
        let parsed = PasswordHash::new(hash).map_err(|_| denied())?;
        Argon2::default()
            .verify_password(password.as_bytes(), &parsed)
            .map_err(|_| denied())?;
        if let Ok(mut cache) = self.verified.write() {
            cache.insert(header.to_string(), role);
        }
        Ok(role)
    }

    #[cfg(test)]
    fn cached(&self) -> usize {
        self.verified.read().map(|c| c.len()).unwrap_or(0)
    }
}

/// Tower layer installing [`AuthService`] in front of the tonic
/// service.
#[derive(Clone)]
pub struct AuthLayer {
    auth: Arc<Authenticator>,
}

impl AuthLayer {
    pub fn new(auth: Arc<Authenticator>) -> Self {
        Self { auth }
    }
}

impl<S> tower::Layer<S> for AuthLayer {
    type Service = AuthService<S>;

    fn layer(&self, inner: S) -> Self::Service {
        AuthService {
            inner,
            auth: self.auth.clone(),
        }
    }
}

/// The single authorization gate: authenticates the header, compares
/// the role against [`minimum_role`] of the request path, and either
/// forwards to the inner service or answers a trailers-only gRPC error
/// (`UNAUTHENTICATED` / `PERMISSION_DENIED`) without running any
/// handler.
#[derive(Clone)]
pub struct AuthService<S> {
    inner: S,
    auth: Arc<Authenticator>,
}

impl<S, ReqBody, ResBody> tower::Service<http::Request<ReqBody>> for AuthService<S>
where
    S: tower::Service<http::Request<ReqBody>, Response = http::Response<ResBody>>,
    ResBody: Default,
{
    type Response = S::Response;
    type Error = S::Error;
    type Future = Either<S::Future, Ready<Result<Self::Response, Self::Error>>>;

    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
        self.inner.poll_ready(cx)
    }

    fn call(&mut self, req: http::Request<ReqBody>) -> Self::Future {
        // The header value itself is a secret and is never logged.
        let header = req
            .headers()
            .get(http::header::AUTHORIZATION)
            .and_then(|value| value.to_str().ok());
        let decision = self.auth.authenticate(header).and_then(|role| {
            let needed = minimum_role(req.uri().path());
            if role >= needed {
                Ok(())
            } else {
                Err(Status::permission_denied(format!(
                    "requires the {} role",
                    needed.name()
                )))
            }
        });
        match decision {
            Ok(()) => Either::Left(self.inner.call(req)),
            Err(status) => Either::Right(ready(Ok(status.into_http()))),
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use argon2::password_hash::{rand_core::OsRng, SaltString};
    use argon2::PasswordHasher;

    /// A PHC hash of `password` with cheap test parameters (the params
    /// travel inside the PHC string, so default verification reads
    /// them back).
    fn hash(password: &str) -> String {
        let params = argon2::Params::new(8, 1, 1, None).expect("params");
        let argon2 = Argon2::new(argon2::Algorithm::Argon2id, argon2::Version::V0x13, params);
        argon2
            .hash_password(password.as_bytes(), &SaltString::generate(&mut OsRng))
            .expect("hash")
            .to_string()
    }

    fn basic(user: &str, password: &str) -> String {
        let encoded =
            base64::engine::general_purpose::STANDARD.encode(format!("{user}:{password}"));
        format!("Basic {encoded}")
    }

    fn authenticator() -> Authenticator {
        Authenticator::new(&AuthSettings {
            owner: Some(hash("os")),
            queue_owner: Some(hash("qos")),
            queue_appender: None,
        })
    }

    // ---- the rights matrix -------------------------------------------

    #[test]
    fn the_method_table_pins_every_rpc_of_the_service() {
        let appender = [
            "Init",
            "GetLibraryNode",
            "GetUpdateStream",
            "Append",
            "CreateLibraryNode",
        ];
        let queue_owner = [
            "Queue",
            "Replace",
            "Remove",
            "Insert",
            "ClearQueue",
            "SetCurrent",
            "ToggleShuffle",
            "ToggleRepeat",
            "TogglePlay",
            "Stop",
            "ChangeVolume",
            "ToggleMute",
            "Next",
            "Prev",
            "RestartTrack",
        ];
        let owner = [
            "CaptureLibraryNode",
            "SaveQueue",
            "RenameLibraryNode",
            "DeleteLibraryNode",
        ];
        // The full service, from crabidy.proto — 24 methods. A new RPC
        // must be added to exactly one list (and the layer keeps it
        // owner-only until then).
        assert_eq!(appender.len() + queue_owner.len() + owner.len(), 24);
        for method in appender {
            let path = format!("{SERVICE_PREFIX}{method}");
            assert_eq!(minimum_role(&path), Role::QueueAppender, "{method}");
        }
        for method in queue_owner {
            let path = format!("{SERVICE_PREFIX}{method}");
            assert_eq!(minimum_role(&path), Role::QueueOwner, "{method}");
        }
        for method in owner {
            let path = format!("{SERVICE_PREFIX}{method}");
            assert_eq!(minimum_role(&path), Role::Owner, "{method}");
        }
    }

    #[test]
    fn unknown_methods_and_foreign_services_are_owner_only() {
        assert_eq!(
            minimum_role("/crabidy.v1.CrabidyService/BrandNewRpc"),
            Role::Owner
        );
        assert_eq!(minimum_role("/grpc.health.v1.Health/Check"), Role::Owner);
        assert_eq!(minimum_role("nonsense"), Role::Owner);
    }

    #[test]
    fn roles_are_ordered_by_privilege() {
        assert!(Role::Owner > Role::QueueOwner);
        assert!(Role::QueueOwner > Role::QueueAppender);
    }

    // ---- authentication ----------------------------------------------

    #[test]
    fn without_configured_hashes_everyone_is_owner() {
        let auth = Authenticator::new(&AuthSettings::default());
        assert!(!auth.enabled());
        assert_eq!(auth.unauthenticated_role(), Some(Role::Owner));
        assert_eq!(auth.authenticate(None).expect("open"), Role::Owner);
    }

    #[test]
    fn the_unauthenticated_role_is_the_highest_unguarded_one() {
        let owner = hash("o");
        let qo = hash("qo");
        let qa = hash("qa");
        // Nothing guarded → owner. Owner guarded → queue-owner. Owner +
        // queue-owner guarded → queue-appender. All guarded → denied.
        let cases = [
            (None, None, None, Some(Role::Owner)),
            (Some(&owner), None, None, Some(Role::QueueOwner)),
            (Some(&owner), Some(&qo), None, Some(Role::QueueAppender)),
            (Some(&owner), Some(&qo), Some(&qa), None),
        ];
        for (owner, queue_owner, queue_appender, expected) in cases {
            let auth = Authenticator::new(&AuthSettings {
                owner: owner.cloned(),
                queue_owner: queue_owner.cloned(),
                queue_appender: queue_appender.cloned(),
            });
            assert_eq!(auth.unauthenticated_role(), expected);
            match expected {
                Some(role) => assert_eq!(auth.authenticate(None).expect("anon"), role),
                None => {
                    let err = auth.authenticate(None).expect_err("locked");
                    assert_eq!(err.code(), tonic::Code::Unauthenticated);
                }
            }
        }
    }

    #[test]
    fn a_header_carrying_bad_credentials_never_falls_back_to_anonymous() {
        // owner guarded, so anonymous callers are queue-owner — but a
        // present-yet-wrong owner login is denied, not downgraded.
        let auth = Authenticator::new(&AuthSettings {
            owner: Some(hash("os")),
            queue_owner: None,
            queue_appender: None,
        });
        assert_eq!(auth.unauthenticated_role(), Some(Role::QueueOwner));
        let err = auth
            .authenticate(Some(&basic("owner", "wrong")))
            .expect_err("wrong password denied");
        assert_eq!(err.code(), tonic::Code::Unauthenticated);
    }

    #[test]
    fn valid_credentials_resolve_their_role() {
        let auth = authenticator();
        assert_eq!(
            auth.authenticate(Some(&basic("owner", "os")))
                .expect("owner"),
            Role::Owner
        );
        assert_eq!(
            auth.authenticate(Some(&basic("queue-owner", "qos")))
                .expect("queue owner"),
            Role::QueueOwner
        );
    }

    #[test]
    fn every_failure_is_the_same_unauthenticated() {
        let auth = authenticator();
        // A missing header is *not* a failure here — it yields the
        // anonymous role. Only present-but-bad headers are failures, and
        // they must be indistinguishable.
        let cases: Vec<Option<String>> = vec![
            Some("Bearer token".to_string()),           // wrong scheme
            Some("Basic !!!not-base64!!!".to_string()), // broken base64
            Some("Basic bm9jb2xvbg==".to_string()),     // no colon
            Some(basic("owner", "wrong")),              // wrong password
            Some(basic("dj", "os")),                    // unknown user
            Some(basic("queue-appender", "anything")),  // role without hash
        ];
        let mut messages = Vec::new();
        for case in &cases {
            let err = auth
                .authenticate(case.as_deref())
                .expect_err(&format!("{case:?}"));
            assert_eq!(err.code(), tonic::Code::Unauthenticated, "{case:?}");
            messages.push(err.message().to_string());
        }
        assert!(
            messages.windows(2).all(|w| w[0] == w[1]),
            "failures must be indistinguishable"
        );
        assert_eq!(auth.cached(), 0, "failures are never cached");
    }

    #[test]
    fn hash_password_output_round_trips_through_the_authenticator() {
        let phc = hash_password("hunter2").expect("hash");
        assert!(phc.starts_with("$argon2id$"), "PHC format");
        let auth = Authenticator::new(&AuthSettings {
            owner: Some(phc),
            queue_owner: None,
            queue_appender: None,
        });
        assert_eq!(
            auth.authenticate(Some(&basic("owner", "hunter2")))
                .expect("round trip"),
            Role::Owner
        );
    }

    #[test]
    fn successful_credentials_are_cached() {
        let auth = authenticator();
        let header = basic("owner", "os");
        assert_eq!(auth.cached(), 0);
        auth.authenticate(Some(&header)).expect("first");
        assert_eq!(auth.cached(), 1);
        auth.authenticate(Some(&header)).expect("cached");
        assert_eq!(auth.cached(), 1, "same credential, one entry");
    }

    // ---- the layer -----------------------------------------------------

    /// Calls the layered service once and returns the response plus
    /// whether the inner service ran.
    fn call_layer(
        auth: Arc<Authenticator>,
        path: &str,
        header: Option<&str>,
    ) -> (http::Response<String>, bool) {
        use std::sync::atomic::{AtomicBool, Ordering};
        use tower::{Layer, Service, ServiceExt};
        let reached = Arc::new(AtomicBool::new(false));
        let flag = reached.clone();
        let inner = tower::service_fn(move |_req: http::Request<()>| {
            flag.store(true, Ordering::SeqCst);
            ready(Ok::<_, std::convert::Infallible>(http::Response::new(
                "handled".to_string(),
            )))
        });
        let mut service = AuthLayer::new(auth).layer(inner);
        let mut req = http::Request::new(());
        *req.uri_mut() = path.parse().expect("uri");
        if let Some(header) = header {
            req.headers_mut().insert(
                http::header::AUTHORIZATION,
                header.parse().expect("header value"),
            );
        }
        let response = futures::executor::block_on(async {
            service.ready().await.expect("ready").call(req).await
        })
        .expect("call");
        (response, reached.load(Ordering::SeqCst))
    }

    fn grpc_status(response: &http::Response<String>) -> Option<&str> {
        response
            .headers()
            .get("grpc-status")
            .and_then(|v| v.to_str().ok())
    }

    #[test]
    fn the_layer_forwards_authorized_requests_only() {
        // owner + queue-owner guarded, appender open → anonymous callers
        // are queue-appenders.
        let auth = Arc::new(authenticator());
        let append = "http://s/crabidy.v1.CrabidyService/Append";
        let capture = "http://s/crabidy.v1.CrabidyService/CaptureLibraryNode";

        // No credentials: the anonymous role (queue-appender) is enough
        // for Append, so the handler runs.
        let (response, reached) = call_layer(auth.clone(), append, None);
        assert!(reached);
        assert_eq!(response.body(), "handled");

        // No credentials against an owner-only method: the anonymous role
        // is insufficient, denied before the handler runs.
        let (response, reached) = call_layer(auth.clone(), capture, None);
        assert!(!reached);
        assert_eq!(grpc_status(&response), Some("7"), "PERMISSION_DENIED");

        // Sufficient role via credentials: forwarded.
        let (response, reached) = call_layer(auth.clone(), capture, Some(&basic("owner", "os")));
        assert!(reached);
        assert_eq!(response.body(), "handled");

        // Valid credentials, insufficient role: denied, handler never
        // runs, and the code distinguishes authorization from
        // authentication.
        let (response, reached) =
            call_layer(auth.clone(), capture, Some(&basic("queue-owner", "qos")));
        assert!(!reached);
        assert_eq!(grpc_status(&response), Some("7"), "PERMISSION_DENIED");

        // Fully locked (every role guarded): an anonymous request is
        // denied with UNAUTHENTICATED before any handler runs.
        let locked = Arc::new(Authenticator::new(&AuthSettings {
            owner: Some(hash("os")),
            queue_owner: Some(hash("qos")),
            queue_appender: Some(hash("aps")),
        }));
        let (response, reached) = call_layer(locked, append, None);
        assert!(!reached);
        assert_eq!(grpc_status(&response), Some("16"), "UNAUTHENTICATED");

        // Auth disabled: everything forwards without a header.
        let open = Arc::new(Authenticator::new(&AuthSettings::default()));
        let (_, reached) = call_layer(open, capture, None);
        assert!(reached);
    }
}