//! `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, list the ones that do, and list the
//! operator group (to `swarm-controller` only, see `permit`). 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,
    /// The one client [`BridgeRequest::ListOperators`] answers.
    operator_reader_client_id: String,
    /// The authelia group whose members are operators.
    operator_group: 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")?)?,
            operator_reader_client_id: env_var("SWARM_AUTHELIA_BRIDGE_OPERATOR_READER_CLIENT_ID")?,
            operator_group: env_var("SWARM_AUTHELIA_BRIDGE_OPERATOR_GROUP")?,
        })
    }
}

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()),
        )
        // This is a systemd-managed daemon — stdout always goes to journald,
        // never a human terminal, and journald doesn't strip ANSI escapes:
        // they land in victorialogs as raw byte-array spam otherwise.
        .with_ansi(false)
        .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 {
    let caller = match authorize(&state, &headers).await {
        Ok(caller) => caller,
        Err(resp) => return resp,
    };
    respond(&state.config, &state.write_lock, caller.as_deref(), req).await
}

/// Everything after authentication: the per-request [`permit`] check, then
/// the operation. Takes the config and the write lock rather than
/// [`AppState`], whose HTTP client only [`authorize`] uses.
async fn respond(
    config: &Config,
    write_lock: &tokio::sync::Mutex<()>,
    caller: Option<&str>,
    req: BridgeRequest,
) -> Response {
    if let Err(refusal) = permit(&req, caller, &config.operator_reader_client_id) {
        // The client id is not a credential; naming it is what lets an
        // operator see which client asked for something it may not have.
        tracing::warn!(client_id = ?caller, request = ?req, "refused a request scoped to another client");
        return refuse(refusal);
    }
    let (op, result) = match req {
        BridgeRequest::EnsureAgentIdentity { name } => {
            ("ensure_identity", handle(config, write_lock, name).await)
        }
        BridgeRequest::ListAgentIdentities => ("list_identities", list(config)),
        BridgeRequest::ListOperators => ("list_operators", list_operators(config)),
    };
    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`]. `Ok` carries the client the
/// token was issued to, for [`permit`].
async fn authorize(state: &AppState, headers: &HeaderMap) -> Result<Option<String>, 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<Option<String>, Refusal> {
    match verdict {
        introspect::Verdict::Active { client_id } => Ok(client_id),
        introspect::Verdict::Inactive => Err(Refusal::Caller),
        introspect::Verdict::Unavailable => Err(Refusal::Bridge),
    }
}

/// Whether an admitted caller may make this particular request.
///
/// [`BridgeRequest::ListOperators`] answers only `operator_reader`, the
/// controller's client: authelia also issues machine tokens to every hive and
/// every hive's agents, and those are as active as the controller's. A caller
/// with no client id is refused it too, rather than compared as a blank.
fn permit(
    req: &BridgeRequest,
    client_id: Option<&str>,
    operator_reader: &str,
) -> Result<(), Refusal> {
    match req {
        BridgeRequest::EnsureAgentIdentity { .. } | BridgeRequest::ListAgentIdentities => Ok(()),
        BridgeRequest::ListOperators if client_id == Some(operator_reader) => Ok(()),
        BridgeRequest::ListOperators => Err(Refusal::Forbidden),
    }
}

/// 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,
    /// The token is valid, but its client may not make this request.
    Forbidden,
}

impl Refusal {
    /// `401` for the caller; `503` for us; `403` for a valid caller asking
    /// for something scoped to another client.
    ///
    /// 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,
            Refusal::Forbidden => StatusCode::FORBIDDEN,
        }
    }

    /// 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",
            Refusal::Forbidden => "not permitted",
        }
    }
}

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

async fn handle(
    cfg: &Config,
    write_lock: &tokio::sync::Mutex<()>,
    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)?;

    // 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 = 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(config: &Config) -> Result<BridgeResponse> {
    let user_store = store::load_store(&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(),
    })
}

/// The operator group's members. No write lock, for [`list`]'s reason.
fn list_operators(config: &Config) -> Result<BridgeResponse> {
    let user_store = store::load_store(&config.users_file)?;
    Ok(BridgeResponse::Operators {
        operators: store::group_members(&user_store, &config.operator_group),
    })
}

/// 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::{
        BridgeRequest, Config, Refusal, StatusCode, classify, introspect::Verdict, permit, respond,
    };

    /// 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 {
                client_id: Some("swarm-controller".to_owned())
            }),
            Ok(Some("swarm-controller".to_owned()))
        );
        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, Refusal::Forbidden] {
            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:?}"
                );
            }
        }
    }

    const CONTROLLER: &str = "swarm-controller";

    /// The trust boundary on the operator list. Every caller below holds a
    /// token authelia calls active — a hive and a hive's agents are issued
    /// machine tokens too — and only the controller's client is answered.
    #[test]
    fn only_the_controller_may_list_operators() {
        let req = BridgeRequest::ListOperators;
        assert_eq!(permit(&req, Some(CONTROLLER), CONTROLLER), Ok(()));
        for other in ["hive-alpha", "hive-alpha-agent", "swarm-authelia-bridge"] {
            assert_eq!(
                permit(&req, Some(other), CONTROLLER),
                Err(Refusal::Forbidden),
                "{other} holds an active token but is not the controller"
            );
        }
        assert_eq!(
            permit(&req, None, CONTROLLER),
            Err(Refusal::Forbidden),
            "a token authelia attributed to no client is not the controller's"
        );
    }

    /// The control for the test above: the agent verbs stay open to any
    /// admitted caller, so the refusal there comes from the scoping and not
    /// from `permit` refusing everyone who is not the controller.
    #[test]
    fn the_agent_verbs_stay_open_to_any_active_caller() {
        for req in [
            BridgeRequest::EnsureAgentIdentity {
                name: "atlas".to_owned(),
            },
            BridgeRequest::ListAgentIdentities,
        ] {
            assert_eq!(
                permit(&req, Some("hive-alpha"), CONTROLLER),
                Ok(()),
                "{req:?}"
            );
            assert_eq!(permit(&req, None, CONTROLLER), Ok(()), "{req:?}");
        }
    }

    #[test]
    fn a_scoped_refusal_is_403_not_401() {
        assert_eq!(
            Refusal::Forbidden.status(),
            StatusCode::FORBIDDEN,
            "the token is valid; a 401 would send the caller off to mint a new one"
        );
    }

    /// A config whose users database holds one operator, `mara`. Every field
    /// the request path does not read is left empty.
    fn config_with_one_operator() -> Config {
        let dir = std::env::temp_dir().join(format!(
            "swarm-authelia-bridge-respond-{}-{}",
            std::process::id(),
            std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .unwrap()
                .as_nanos()
        ));
        std::fs::create_dir_all(&dir).unwrap();
        let users_file = dir.join("users.yml");
        std::fs::write(
            &users_file,
            "users:\n  mara:\n    displayname: mara\n    password: \"$argon2id$x\"\n    groups: [admins]\n",
        )
        .unwrap();
        Config {
            bind: String::new(),
            users_file,
            authelia_bin: std::path::PathBuf::new(),
            introspection_url: String::new(),
            client_id: String::new(),
            client_secret: String::new(),
            operator_reader_client_id: CONTROLLER.to_owned(),
            operator_group: "admins".to_owned(),
        }
    }

    /// The request path, not only [`permit`]: a dispatch that skipped the
    /// check would answer the hive below with the operator list.
    #[tokio::test]
    async fn the_request_path_answers_the_operator_list_to_the_controller_only() {
        let config = config_with_one_operator();
        let lock = tokio::sync::Mutex::new(());
        for caller in [Some("hive-alpha"), None] {
            let refused = respond(&config, &lock, caller, BridgeRequest::ListOperators).await;
            assert_eq!(refused.status(), StatusCode::FORBIDDEN, "{caller:?}");
        }

        let answered = respond(
            &config,
            &lock,
            Some(CONTROLLER),
            BridgeRequest::ListOperators,
        )
        .await;
        assert_eq!(answered.status(), StatusCode::OK);
        let body = axum::body::to_bytes(answered.into_body(), usize::MAX)
            .await
            .unwrap();
        assert_eq!(
            serde_json::from_slice::<serde_json::Value>(&body).unwrap(),
            serde_json::json!({"status": "operators", "operators": ["mara"]})
        );
        let _ = std::fs::remove_dir_all(config.users_file.parent().unwrap());
    }
}