//! `swarm-authelia-bridge` — the only thing allowed to write
//! `swarm-authelia`'s users database.
//!
//! Runs **inside the `swarm-authelia` container**, as
//! `User = "authelia-swarm";` (the literal name
//! `swarm-authelia.nix`'s `unitName` already derives) — the same user
//! authelia's own instance runs as, and therefore the file's actual
//! owner. That is the whole answer to "how does an unprivileged
//! `swarm-controller` touch a file it doesn't own": it doesn't — this
//! process does, sidestepping the uid boundary instead of bridging it
//! with root/`CAP_CHOWN`/a shared group (all examined and rejected — see
//! `swarmctl/README.md`'s own identical analysis of this same file).
//!
//! Network-facing, not unix-socket-only: `swarm-authelia`'s container
//! shares the host netns (`privateNetwork = false`, same as `hive-forge`/
//! `hive-matrix`), so one listener serves a co-located `swarm-controller`
//! (loopback) and a split-host one (bind wider, firewall it) with the
//! same code path — no separate transport for the cross-host case.
//! Bearer-authenticated via authelia's own OIDC token introspection (RFC
//! 7662) against `swarm-controller`'s **existing** machine-client
//! identity (already minted for the queue connection) — no new
//! credential to mint or deliver, and always local/fast to verify since
//! this process runs right next to the authelia it asks.
//!
//! One endpoint, `POST /requests`, body =
//! [`swarm_authelia_bridge_sock::BridgeRequest`] verbatim: ensure an agent
//! exists as an authelia subject, and list the ones that do. Not a
//! wholesale-replace-the-file API — this process owns rendering
//! `users.yml` internally; see `store` module.

mod introspect;
mod store;

use std::sync::Arc;

use anyhow::{Context, Result, bail};
use axum::extract::State;
use axum::http::{HeaderMap, StatusCode};
use axum::response::{IntoResponse, Response};
use axum::routing::post;
use axum::{Json, Router};
use swarm_authelia_bridge_sock::{AgentIdentity, BridgeRequest, BridgeResponse};

/// Where the introspecting bearer token is authenticated *to* — this
/// bridge's own authelia machine client, distinct from
/// `swarm-controller`'s (the token being checked). Set by the nix module
/// that provisions this bridge's own client alongside authelia.
struct Config {
    bind: String,
    users_file: std::path::PathBuf,
    authelia_bin: std::path::PathBuf,
    introspection_url: String,
    client_id: String,
    client_secret: String,
}

impl Config {
    fn from_env() -> Result<Self> {
        Ok(Self {
            bind: env_var("SWARM_AUTHELIA_BRIDGE_BIND")?,
            users_file: env_var("SWARM_AUTHELIA_BRIDGE_USERS_FILE")?.into(),
            authelia_bin: env_var("SWARM_AUTHELIA_BRIDGE_AUTHELIA_BIN")?.into(),
            introspection_url: env_var("SWARM_AUTHELIA_BRIDGE_INTROSPECTION_URL")?,
            client_id: env_var("SWARM_AUTHELIA_BRIDGE_CLIENT_ID")?,
            client_secret: read_secret_file(&env_var("SWARM_AUTHELIA_BRIDGE_CLIENT_SECRET_FILE")?)?,
        })
    }
}

fn env_var(name: &str) -> Result<String> {
    std::env::var(name).with_context(|| format!("{name} is unset"))
}

fn read_secret_file(path: &str) -> Result<String> {
    let raw =
        std::fs::read_to_string(path).with_context(|| format!("reading secret file {path}"))?;
    let trimmed = raw.trim();
    if trimmed.is_empty() {
        bail!("secret file {path} is empty");
    }
    Ok(trimmed.to_owned())
}

struct AppState {
    config: Config,
    http: reqwest::Client,
    /// Serializes `handle`'s load → insert → publish sequence. Without this,
    /// two `EnsureAgentIdentity` requests landing close together (real: the
    /// controller's job worker claims and spawns nodes without waiting for
    /// each to finish, and `SwarmResourceKind` declares no resource dep
    /// between two `CreateIdentity` jobs, so they run concurrently) both
    /// `load_store` the same snapshot, both insert their own agent, and
    /// whichever `publish`es second silently drops the first agent's entry —
    /// the store is a plain file, not a database with its own concurrency
    /// control. `tokio::sync::Mutex`, not `std`'s: held across the `.await`s
    /// in `generate_password` and `publish`.
    write_lock: tokio::sync::Mutex<()>,
}

#[tokio::main]
async fn main() -> Result<()> {
    tracing_subscriber::fmt()
        .with_env_filter(
            tracing_subscriber::EnvFilter::try_from_default_env().unwrap_or_else(|_| "info".into()),
        )
        .init();

    let config = Config::from_env()?;
    let bind = config.bind.clone();
    let state = Arc::new(AppState {
        config,
        http: reqwest::Client::new(),
        write_lock: tokio::sync::Mutex::new(()),
    });

    let app = Router::new()
        .route("/requests", post(handle_request))
        .with_state(state);

    let listener = tokio::net::TcpListener::bind(&bind)
        .await
        .with_context(|| format!("binding {bind}"))?;
    tracing::info!(%bind, "swarm-authelia-bridge listening");
    axum::serve(listener, app)
        .await
        .context("serving swarm-authelia-bridge")
}

/// The single endpoint — body is the wire-crate's [`BridgeRequest`]
/// verbatim (JSON), not a per-operation REST route. That is what let the
/// second operation arrive as a match arm rather than as another route
/// with its own extractor and its own copy of the authorize call.
async fn handle_request(
    State(state): State<Arc<AppState>>,
    headers: HeaderMap,
    Json(req): Json<BridgeRequest>,
) -> Response {
    if let Err(resp) = authorize(&state, &headers).await {
        return resp;
    }
    let (op, result) = match req {
        BridgeRequest::EnsureAgentIdentity { name } => {
            ("ensure_identity", handle(&state, name).await)
        }
        BridgeRequest::ListAgentIdentities => ("list_identities", list(&state)),
    };
    match result {
        Ok(body) => (StatusCode::OK, Json(body)).into_response(),
        Err(e) => {
            let detail = format!("{e:#}");
            tracing::warn!(op, error = %detail, "request failed");
            (
                StatusCode::INTERNAL_SERVER_ERROR,
                Json(BridgeResponse::Error { message: detail }),
            )
                .into_response()
        }
    }
}

/// Extract + introspect the bearer token. `Err` carries the response to
/// return outright — kept separate from `handle`'s error path, which is
/// "the op itself failed," a different case from "the caller was never let
/// in."
///
/// Anything short of an active token is refused; **which status** depends
/// on whose credential failed, see [`Refusal`].
async fn authorize(state: &AppState, headers: &HeaderMap) -> Result<(), Response> {
    let Some(auth) = headers.get(axum::http::header::AUTHORIZATION) else {
        return Err(refuse(Refusal::Caller));
    };
    let Ok(auth) = auth.to_str() else {
        return Err(refuse(Refusal::Caller));
    };
    let Some(token) = auth.strip_prefix("Bearer ") else {
        return Err(refuse(Refusal::Caller));
    };
    let verdict = introspect::introspect(
        &state.http,
        &state.config.introspection_url,
        &state.config.client_id,
        &state.config.client_secret,
        token,
    )
    .await;
    classify(verdict).map_err(refuse)
}

/// The admission decision, and the only place that makes it.
///
/// Fail-closed: `Ok(())` is reachable from exactly one variant. Split out
/// of [`authorize`] so the rule is testable without an authelia — and it is
/// the *real* rule, not a restatement of it, because this is the function
/// the request path calls.
fn classify(verdict: introspect::Verdict) -> Result<(), Refusal> {
    match verdict {
        introspect::Verdict::Active => Ok(()),
        introspect::Verdict::Inactive => Err(Refusal::Caller),
        introspect::Verdict::Unavailable => Err(Refusal::Bridge),
    }
}

/// Whose credential was at fault when a request was refused.
///
/// The distinction is the whole point: these used to be one 401, so a
/// caller whose token was fine was told its token was invalid whenever
/// *this bridge* could not authenticate itself to authelia.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum Refusal {
    /// No token, an unusable header, or a token authelia rejected.
    Caller,
    /// This bridge could not get an answer out of authelia at all, so the
    /// request was never evaluated.
    Bridge,
}

impl Refusal {
    /// `401` for the caller; `503` for us.
    ///
    /// 401 would be wrong for [`Refusal::Bridge`] under RFC 7235 — nothing
    /// about the caller's credential was determined — and it is also the
    /// least useful thing to say, because a client that believes its token
    /// is bad will go and mint a new one, which cannot help.
    ///
    /// 503 rather than 500 because the overwhelmingly common cause is
    /// transient by construction: this bridge's own secret is minted on
    /// authelia's first boot and delivered by a separate unit, so the
    /// window where it is absent closes on its own. A client retrying is
    /// doing the right thing.
    fn status(self) -> StatusCode {
        match self {
            Refusal::Caller => StatusCode::UNAUTHORIZED,
            Refusal::Bridge => StatusCode::SERVICE_UNAVAILABLE,
        }
    }

    /// Deliberately coarse for [`Refusal::Caller`]: an unauthenticated
    /// caller learns that it was refused, never which of the four ways.
    /// The `Bridge` message says only that validation could not happen —
    /// no status code, no upstream detail, nothing about the secret. The
    /// discriminating detail is in this bridge's journal, where it belongs.
    fn message(self) -> &'static str {
        match self {
            Refusal::Caller => "missing or invalid bearer token",
            Refusal::Bridge => "cannot validate credentials",
        }
    }
}

fn refuse(refusal: Refusal) -> Response {
    (
        refusal.status(),
        Json(BridgeResponse::Error {
            message: refusal.message().to_owned(),
        }),
    )
        .into_response()
}

async fn handle(state: &AppState, name: String) -> Result<BridgeResponse> {
    store::validate_username(&name)?;
    // After the charset check and before anything touches the store: a
    // reserved name is refused on its own terms, not as a side effect of a
    // collision with whatever happens to be in the file today.
    store::reject_reserved_name(&name)?;
    let cfg = &state.config;

    // Held across the whole load → insert → publish sequence, not just the
    // write — two concurrent `EnsureAgentIdentity` calls must not both read
    // the same on-disk snapshot before either publishes. See the field's
    // doc comment on `AppState::write_lock` for why this is reachable in
    // practice, not just theoretically.
    let _write_guard = state.write_lock.lock().await;

    let mut user_store = store::load_store(&cfg.users_file)?;
    if user_store.users.contains_key(&name) {
        // The subject is there, so no password is minted. Whether it gained
        // the marker here is the caller's business, not an implementation
        // detail: an unmarked subject is either a pre-marker agent (the
        // intended migration) or a human created without a group, and this
        // process cannot tell them apart. So the two outcomes are reported
        // as different variants rather than folded into one.
        if store::mark_as_agent(&mut user_store, &name) {
            store::publish(&cfg.users_file, &mut user_store)?;
            // `warn`, not `info`: the uneventful path is silent, so anything
            // logged here is a subject that was *not* an agent a moment ago.
            // Names the account because "which one" is the entire question
            // an operator will have.
            tracing::warn!(
                subject = %name,
                "added the agent marker to an EXISTING identity — intended if this \
                 predates the marker, but the store cannot distinguish that from a \
                 human account created without a group"
            );
            return Ok(BridgeResponse::Healed);
        }
        return Ok(BridgeResponse::AlreadyExists);
    }
    let generated = generate_password(&cfg.authelia_bin).await?;
    user_store.users.insert(
        name.clone(),
        store::User {
            displayname: name.clone(),
            password: generated,
            email: None,
            groups: vec![store::AGENT_GROUP.to_owned()],
            extra: std::collections::BTreeMap::new(),
        },
    );
    store::publish(&cfg.users_file, &mut user_store)?;
    tracing::info!(agent = %name, "created authelia identity");
    Ok(BridgeResponse::Created)
}

/// The roster: every subject carrying the agent marker.
///
/// Takes no write lock. The store is replaced by an atomic rename, so a
/// reader sees either the whole old file or the whole new one — and holding
/// the lock would make a read wait behind a password mint, which is an
/// external process.
fn list(state: &AppState) -> Result<BridgeResponse> {
    let user_store = store::load_store(&state.config.users_file)?;
    Ok(BridgeResponse::Agents {
        agents: store::agent_names(&user_store)
            .into_iter()
            .filter_map(|name| match hive_types::Ident::parse(&name) {
                Ok(ident) => Some(AgentIdentity { name: ident }),
                // The store's charset is wider than an agent name's, because
                // it holds humans too. Every agent this bridge creates is a
                // legal `Ident` by construction, so reaching here means a
                // subject was hand-added to the agent group that cannot be an
                // agent. Skipped rather than failing the whole roster — one
                // bad row must not make every other agent invisible — but
                // logged, because silently shrinking an answer is how a
                // roster starts lying about who is missing.
                Err(reason) => {
                    tracing::warn!(
                        subject = %name,
                        %reason,
                        "subject carries the agent group but is not a legal agent name; \
                         omitting it from the roster"
                    );
                    None
                }
            })
            .collect(),
    })
}

/// Mint a password digest via the **configured** authelia — the argon2
/// parameters baked into a hash have to match the verifier's, same
/// reasoning `swarmctl::generate_password` documents (this is
/// deliberately a fresh async port of that function, not a shared one —
/// `swarmctl` is a separate binary crate with no lib target).
///
/// `--random`, not `--password <pw>`: `/proc/<pid>/cmdline` is
/// world-readable, so a password on argv would be readable by any local
/// process for the call's lifetime. Only the digest is kept — the
/// plaintext is generated and immediately discarded, since nothing reads
/// it back today (still an open question what an agent's login actually
/// uses this for).
async fn generate_password(bin: &std::path::Path) -> Result<String> {
    let out = tokio::process::Command::new(bin)
        .args(["crypto", "hash", "generate", "argon2", "--random"])
        .output()
        .await
        .with_context(|| format!("running {}", bin.display()))?;
    if !out.status.success() {
        bail!(
            "{} failed ({}): {}",
            bin.display(),
            out.status,
            String::from_utf8_lossy(&out.stderr).trim()
        );
    }
    let stdout = String::from_utf8(out.stdout).context("authelia printed non-UTF-8 output")?;
    stdout
        .lines()
        .find_map(|line| line.trim().strip_prefix("Digest:"))
        .map(|v| v.trim().to_owned())
        .filter(|v| !v.is_empty())
        .with_context(|| {
            format!(
                "could not parse {}'s output: missing 'Digest:'",
                bin.display()
            )
        })
}

#[cfg(test)]
mod tests {
    use super::{Refusal, StatusCode, classify, introspect::Verdict};

    /// Fail-closed, asserted per variant rather than as one negation:
    /// splitting blame out of a single denial is exactly the change that
    /// could have widened admission, so each refusing variant is named on
    /// its own line.
    #[test]
    fn only_an_active_token_is_admitted() {
        assert_eq!(classify(Verdict::Active), Ok(()));
        assert_eq!(
            classify(Verdict::Inactive),
            Err(Refusal::Caller),
            "authelia evaluated the token and said no"
        );
        assert_eq!(
            classify(Verdict::Unavailable),
            Err(Refusal::Bridge),
            "a bridge that cannot ask must not admit — this variant reports a \
             different fault, it is not a softer denial"
        );
    }

    /// The whole point of the split, asserted on the value rather than
    /// on message text — the previous code picked its status by matching on
    /// a log message's wording, which is what let the two collapse.
    #[test]
    fn a_refusal_names_whose_credential_failed() {
        assert_eq!(
            Refusal::Caller.status(),
            StatusCode::UNAUTHORIZED,
            "the caller's token is the problem; retrying with the same one will not help"
        );
        assert_eq!(
            Refusal::Bridge.status(),
            StatusCode::SERVICE_UNAVAILABLE,
            "the request was never evaluated — 401 here accuses the caller of our fault"
        );
        assert_ne!(
            Refusal::Caller.status(),
            Refusal::Bridge.status(),
            "collapsing these back into one status is the bug this exists to prevent"
        );
    }

    /// No response body may leak which of the four caller-side causes
    /// applied, nor anything about this bridge's own credential.
    #[test]
    fn refusal_messages_say_nothing_useful_to_an_attacker() {
        for refusal in [Refusal::Caller, Refusal::Bridge] {
            let message = refusal.message();
            for leak in ["secret", "client_id", "authelia", "http", "status"] {
                assert!(
                    !message.to_ascii_lowercase().contains(leak),
                    "{refusal:?} message leaks {leak:?}: {message:?}"
                );
            }
        }
    }
}