//! End-to-end: pull and push mirrors against one real `forge` binary, talking to itself over
//! HTTP smart git. The instance allows plain HTTP and allow-lists `127.0.0.1` in `[webhooks]`, so
//! its own mirror worker may reach the loopback listener. Nothing else about the SSRF guard is
//! relaxed.
//!
//! Scenarios:
//! - pull mirror `dst` of `alice/src` (heads and tags) created through the new-repo form, then a
//!   second change picked up by "Sync now"; a push to `dst` is refused;
//! - push mirror `pm` to `alice/target` (created empty) after a push to `pm`;
//! - the mirror pages and the server log never contain the access token.

mod common;
use common::{BIN, Server, launch_with, pick_ports, tool_on_path, write_config_with};
use std::collections::BTreeSet;
use std::fs;
use std::path::{Path, PathBuf};
use std::process::{Command, Output};
use std::time::{Duration, Instant};

use reqwest::header::{LOCATION, SET_COOKIE};

const ALICE_PW: &str = "alice-password-123";
const WEBHOOK_TOML: &str = "\n[webhooks]\nallow_http = true\nallowed_hosts = [\"127.0.0.1\"]\n";
const SYNC_TIMEOUT: Duration = Duration::from_secs(30);
/// Push mirrors run after the 10s debounce, but the worker only re-checks on its 60s poll (or a
/// wake-up, which fires before the debounce expires). Allow poll plus debounce plus slack.
const PUSH_TIMEOUT: Duration = Duration::from_secs(90);

struct Resp {
    status: u16,
    location: Option<String>,
    set_cookies: Vec<String>,
    body: String,
}

struct Http {
    client: reqwest::Client,
    base: String,
}

impl Http {
    fn new(port: u16) -> Self {
        let client = reqwest::Client::builder()
            .redirect(reqwest::redirect::Policy::none())
            .timeout(Duration::from_secs(60))
            .build()
            .expect("http client");
        Self {
            client,
            base: format!("http://127.0.0.1:{port}"),
        }
    }

    async fn get_cookie(&self, path: &str, cookie: &str) -> Resp {
        let req = self
            .client
            .get(format!("{}{path}", self.base))
            .header(reqwest::header::COOKIE, cookie);
        send(req).await
    }

    async fn post_form(&self, path: &str, form: &[(&str, &str)], cookie: Option<&str>) -> Resp {
        let mut req = self.client.post(format!("{}{path}", self.base)).form(form);
        if let Some(cookie) = cookie {
            req = req.header(reqwest::header::COOKIE, cookie);
        }
        send(req).await
    }
}

async fn send(req: reqwest::RequestBuilder) -> Resp {
    let res = req.send().await.expect("http request");
    let status = res.status().as_u16();
    let location = res
        .headers()
        .get(LOCATION)
        .and_then(|v| v.to_str().ok())
        .map(str::to_owned);
    let set_cookies = res
        .headers()
        .get_all(SET_COOKIE)
        .iter()
        .filter_map(|v| v.to_str().ok().map(str::to_owned))
        .collect();
    let body = res.text().await.unwrap_or_default();
    Resp {
        status,
        location,
        set_cookies,
        body,
    }
}

/// Returns `name=value` for the cookie whose name ends with `suffix`. Cleared cookies are ignored.
fn cookie_named(resp: &Resp, suffix: &str) -> Option<String> {
    resp.set_cookies.iter().find_map(|set| {
        let pair = set.split(';').next()?;
        let (name, value) = pair.split_once('=')?;
        (name.ends_with(suffix) && !value.is_empty()).then(|| format!("{name}={value}"))
    })
}

/// The value of the first `<input name="...">` on a page.
fn input_value(body: &str, name: &str) -> String {
    let doc = scraper::Html::parse_document(body);
    let selector = scraper::Selector::parse(&format!("input[name=\"{name}\"]")).expect("selector");
    doc.select(&selector)
        .next()
        .and_then(|el| el.value().attr("value"))
        .unwrap_or_else(|| panic!("no input named {name} on page"))
        .to_owned()
}

/// The `action` of every form whose action ends with `suffix`, in page order.
fn form_actions(body: &str, suffix: &str) -> Vec<String> {
    let doc = scraper::Html::parse_document(body);
    let selector = scraper::Selector::parse("form[action]").expect("selector");
    doc.select(&selector)
        .filter_map(|el| el.value().attr("action"))
        .filter(|action| action.ends_with(suffix))
        .map(str::to_owned)
        .collect()
}

fn run_admin(config: &Path, username: &str, extra: &[&str], password: &str) {
    let out = Command::new(BIN)
        .arg("--config")
        .arg(config)
        .args(["admin", "create-user", username])
        .args(extra)
        .env("FORGE_PASSWORD", password)
        .output()
        .expect("run admin create-user");
    assert!(
        out.status.success(),
        "create-user {username} failed:\n{}",
        String::from_utf8_lossy(&out.stderr)
    );
}

/// Runs git with credential helpers disabled and extra env. Returns the raw output.
fn git_output(cwd: &Path, args: &[&str], env: &[(&str, &str)]) -> Output {
    let mut cmd = Command::new("git");
    cmd.current_dir(cwd)
        .args(["-c", "credential.helper="])
        .args(args)
        .env("GIT_TERMINAL_PROMPT", "0");
    for (key, value) in env {
        cmd.env(key, value);
    }
    cmd.output().expect("run git")
}

/// Runs git, returning stdout trimmed. Panics on failure.
fn git_env(cwd: &Path, args: &[&str], env: &[(&str, &str)]) -> String {
    let out = git_output(cwd, args, env);
    assert!(
        out.status.success(),
        "git {args:?} failed:\n{}",
        String::from_utf8_lossy(&out.stderr)
    );
    String::from_utf8_lossy(&out.stdout).trim().to_owned()
}

fn git(cwd: &Path, args: &[&str]) -> String {
    git_env(cwd, args, &[])
}

/// Writes `file`, commits it as `who` and returns the new HEAD.
fn commit(repo: &Path, file: &str, contents: &str, message: &str) -> String {
    fs::write(repo.join(file), contents).expect("write file");
    git(repo, &["add", file]);
    git_env(
        repo,
        &["-c", "commit.gpgsign=false", "commit", "-q", "-m", message],
        &[
            ("GIT_AUTHOR_NAME", "alice"),
            ("GIT_AUTHOR_EMAIL", "alice@example.test"),
            ("GIT_COMMITTER_NAME", "alice"),
            ("GIT_COMMITTER_EMAIL", "alice@example.test"),
        ],
    );
    git(repo, &["rev-parse", "HEAD"])
}

/// Clones `url` into `into` (a path that does not exist yet) and returns it.
fn clone(url: &str, into: &Path) -> PathBuf {
    let parent = into.parent().expect("clone target has a parent");
    git(
        parent,
        &["clone", "-q", url, into.to_str().expect("utf-8 path")],
    );
    into.to_path_buf()
}

/// The HTTPS remote for `owner/repo`, authenticated with a token.
fn remote(port: u16, token: &str, owner: &str, repo: &str) -> String {
    format!("http://alice:{token}@127.0.0.1:{port}/{owner}/{repo}.git")
}

/// The heads and tags a remote advertises, as sorted `sha refname` lines. HEAD is left out.
/// Returns `None` if the remote cannot be listed.
fn branches_and_tags(url: &str) -> Option<BTreeSet<String>> {
    let out = git_output(&std::env::temp_dir(), &["ls-remote", url], &[]);
    if !out.status.success() {
        return None;
    }
    let text = String::from_utf8_lossy(&out.stdout);
    Some(
        text.lines()
            .filter(|line| line.contains("refs/heads/") || line.contains("refs/tags/"))
            .map(str::to_owned)
            .collect(),
    )
}

/// The sha a remote branch points at, if the branch exists.
fn remote_head(url: &str, branch: &str) -> Option<String> {
    let out = git_output(
        &std::env::temp_dir(),
        &["ls-remote", url, &format!("refs/heads/{branch}")],
        &[],
    );
    if !out.status.success() {
        return None;
    }
    String::from_utf8_lossy(&out.stdout)
        .split_whitespace()
        .next()
        .map(str::to_owned)
}

/// Polls `check` every 200ms until it returns true, failing after `SYNC_TIMEOUT`.
async fn wait_for(what: &str, check: impl Fn() -> bool) {
    let deadline = Instant::now() + SYNC_TIMEOUT;
    while !check() {
        assert!(
            Instant::now() < deadline,
            "timed out after {SYNC_TIMEOUT:?} waiting for {what}"
        );
        tokio::time::sleep(Duration::from_millis(200)).await;
    }
}

/// Logs in through the real `/login` form and returns the session cookie header.
async fn login(http: &Http, username: &str, password: &str) -> String {
    let page = http.get_cookie("/login", "").await;
    assert_eq!(page.status, 200, "GET /login");
    let precsrf_cookie = cookie_named(&page, "forge_precsrf").expect("precsrf cookie");
    let precsrf = input_value(&page.body, "precsrf");
    let res = http
        .post_form(
            "/login",
            &[
                ("username", username),
                ("password", password),
                ("next", "/"),
                ("precsrf", &precsrf),
            ],
            Some(&precsrf_cookie),
        )
        .await;
    assert_eq!(res.status, 303, "login as {username}: {}", res.body);
    cookie_named(&res, "forge_session").expect("session cookie after login")
}

/// The session's CSRF token, scraped from the token settings page.
async fn csrf_token(http: &Http, session: &str) -> String {
    let page = http.get_cookie("/settings/tokens", session).await;
    assert_eq!(page.status, 200, "GET /settings/tokens");
    input_value(&page.body, "csrf")
}

/// Creates an access token through `/settings/tokens` and returns the token shown once.
async fn create_token(http: &Http, session: &str, name: &str) -> String {
    let csrf = csrf_token(http, session).await;
    let res = http
        .post_form(
            "/settings/tokens",
            &[("csrf", &csrf), ("name", name), ("expires", "30")],
            Some(session),
        )
        .await;
    assert_eq!(res.status, 200, "create token: {}", res.body);
    let doc = scraper::Html::parse_document(&res.body);
    let selector = scraper::Selector::parse("pre.code code").expect("selector");
    let token: String = doc
        .select(&selector)
        .next()
        .expect("token shown after creation")
        .text()
        .collect();
    let token = token.trim().to_owned();
    assert!(token.starts_with("forge_"), "unexpected token shape");
    token
}

/// Submits the new-repo form. `mirror` adds the "Mirror an existing repository" fields.
async fn new_repo(http: &Http, session: &str, name: &str, mirror: Option<(&str, &str, &str)>) {
    let page = http.get_cookie("/new", session).await;
    assert_eq!(page.status, 200, "GET /new");
    let csrf = input_value(&page.body, "csrf");
    let mut form = vec![
        ("csrf", csrf.as_str()),
        ("name", name),
        ("description", ""),
        ("visibility", "public"),
        ("default_branch", "main"),
    ];
    if let Some((url, username, secret)) = mirror {
        form.extend([
            ("mirror_url", url),
            ("mirror_username", username),
            ("mirror_secret", secret),
            ("mirror_interval", "60"),
        ]);
    }
    let res = http.post_form("/new", &form, Some(session)).await;
    assert_eq!(res.status, 303, "create repo {name}: {}", res.body);
    assert_eq!(
        res.location.as_deref(),
        Some(format!("/alice/{name}").as_str())
    );
}

/// Adds a push mirror to `alice/repo` through its settings page.
async fn add_push_mirror(http: &Http, session: &str, repo: &str, url: &str, secret: &str) {
    let path = format!("/alice/{repo}/settings/mirrors");
    let page = http.get_cookie(&path, session).await;
    assert_eq!(page.status, 200, "GET {path}");
    let csrf = input_value(&page.body, "csrf");
    let res = http
        .post_form(
            &path,
            &[
                ("csrf", &csrf),
                ("url", url),
                ("username", "alice"),
                ("secret", secret),
                ("interval", "60"),
            ],
            Some(session),
        )
        .await;
    assert_eq!(res.status, 303, "add push mirror: {}", res.body);
}

/// Clicks "Sync now" on the first mirror of `alice/repo`.
async fn sync_now(http: &Http, session: &str, repo: &str) {
    let path = format!("/alice/{repo}/settings/mirrors");
    let page = http.get_cookie(&path, session).await;
    assert_eq!(page.status, 200, "GET {path}");
    let action = form_actions(&page.body, "/sync")
        .into_iter()
        .next()
        .expect("a sync form on the mirrors page");
    let csrf = input_value(&page.body, "csrf");
    let res = http
        .post_form(&action, &[("csrf", &csrf)], Some(session))
        .await;
    assert_eq!(res.status, 303, "sync now: {}", res.body);
}

/// A running server with its own temp data dir. Field order matters: the server is dropped (and
/// killed) before the temp dir is removed.
struct Booted {
    _server: Server,
    _root: tempfile::TempDir,
    dir: PathBuf,
    http: Http,
    http_port: u16,
}

/// Creates alice (admin) and boots the server with plain HTTP and loopback mirrors allowed.
async fn boot() -> Booted {
    if !tool_on_path("git") {
        panic!("`git` was not found on PATH");
    }
    // Keep the path short: the internal unix socket path is limited to 108 bytes.
    let root = tempfile::Builder::new()
        .prefix("mr2e")
        .tempdir_in("/tmp")
        .expect("temp dir");
    let dir = root.path().to_path_buf();
    let ports = pick_ports();
    let config = write_config_with(&dir, ports, WEBHOOK_TOML);

    run_admin(&config, "alice", &["--admin"], ALICE_PW);
    let (server, ports) = launch_with(&dir, ports, WEBHOOK_TOML).await;
    Booted {
        _server: server,
        _root: root,
        dir,
        http: Http::new(ports.http),
        http_port: ports.http,
    }
}

/// The server log, for assertion messages and the token leak check.
fn server_log(b: &Booted) -> String {
    fs::read_to_string(b.dir.join("serve.log")).unwrap_or_default()
}

#[tokio::test]
async fn pull_and_push_mirrors_sync_over_smart_http() {
    let b = boot().await;
    let session = login(&b.http, "alice", ALICE_PW).await;
    let token = create_token(&b.http, &session, "mirror-token").await;
    let port = b.http_port;

    // Source repo with a commit on main, a second branch and an annotated tag.
    new_repo(&b.http, &session, "src", None).await;
    let src_work = clone(
        &remote(port, &token, "alice", "src"),
        &b.dir.join("src-work"),
    );
    let first = commit(&src_work, "base.txt", "base\n", "base commit");
    git(&src_work, &["push", "-q", "origin", "HEAD:refs/heads/main"]);
    git(&src_work, &["tag", "-a", "v1", "-m", "release one"]);
    git(&src_work, &["push", "-q", "origin", "refs/tags/v1"]);
    git(
        &src_work,
        &["push", "-q", "origin", "HEAD:refs/heads/feature"],
    );
    let src_url = format!("http://127.0.0.1:{port}/alice/src.git");
    let src_refs = branches_and_tags(&remote(port, &token, "alice", "src")).expect("list src");
    assert!(src_refs.iter().any(|l| l.ends_with("refs/tags/v1")));
    assert!(src_refs.iter().any(|l| l.ends_with("refs/heads/feature")));

    // Pull mirror created through the new-repo form. Creation schedules the first sync now.
    new_repo(&b.http, &session, "dst", Some((&src_url, "alice", &token))).await;
    let dst_url = remote(port, &token, "alice", "dst");
    wait_for("dst to hold src's heads and tags", || {
        branches_and_tags(&dst_url).is_some_and(|refs| refs == src_refs)
    })
    .await;
    assert_eq!(
        remote_head(&dst_url, "main").as_deref(),
        Some(first.as_str())
    );

    // A push to the pull mirror is refused by pre-receive.
    let dst_work = clone(&dst_url, &b.dir.join("dst-work"));
    commit(&dst_work, "local.txt", "local\n", "local change");
    let refused = git_output(&dst_work, &["push", "origin", "HEAD:refs/heads/main"], &[]);
    assert!(
        !refused.status.success(),
        "push to a pull mirror must be rejected"
    );
    let stderr = String::from_utf8_lossy(&refused.stderr);
    assert!(
        stderr.contains("pull mirror"),
        "rejection should name the pull mirror:\n{stderr}"
    );

    // A new commit on src, picked up by "Sync now" on dst.
    let second = commit(&src_work, "next.txt", "next\n", "second commit");
    git(&src_work, &["push", "-q", "origin", "HEAD:refs/heads/main"]);
    sync_now(&b.http, &session, "dst").await;
    wait_for("dst main to follow src after sync now", || {
        remote_head(&dst_url, "main").as_deref() == Some(second.as_str())
    })
    .await;

    // Push mirror: an empty target repo receives pm's commit.
    new_repo(&b.http, &session, "target", None).await;
    new_repo(&b.http, &session, "pm", None).await;
    let target_url = format!("http://127.0.0.1:{port}/alice/target.git");
    add_push_mirror(&b.http, &session, "pm", &target_url, &token).await;
    let pm_work = b.dir.join("pm-work");
    fs::create_dir_all(&pm_work).expect("pm work dir");
    git(&pm_work, &["init", "-q", "-b", "main"]);
    let pushed = commit(&pm_work, "pm.txt", "pm\n", "pm commit");
    git(
        &pm_work,
        &[
            "push",
            "-q",
            &remote(port, &token, "alice", "pm"),
            "HEAD:refs/heads/main",
        ],
    );
    let target_remote = remote(port, &token, "alice", "target");
    let delivered = {
        let started = Instant::now();
        let deadline = started + PUSH_TIMEOUT;
        loop {
            if remote_head(&target_remote, "main").as_deref() == Some(pushed.as_str()) {
                eprintln!("push mirror delivered after {:?}", started.elapsed());
                break true;
            }
            if Instant::now() >= deadline {
                break false;
            }
            tokio::time::sleep(Duration::from_millis(200)).await;
        }
    };
    if !delivered {
        let page = b
            .http
            .get_cookie("/alice/pm/settings/mirrors", &session)
            .await;
        panic!(
            "push mirror did not deliver {pushed} to target within {PUSH_TIMEOUT:?}\n\
             pm mirrors page status {}:\n{}\nserver log:\n{}",
            page.status,
            page.body,
            server_log(&b)
        );
    }

    // The token never shows up in the mirror pages or the server log.
    for repo in ["dst", "pm"] {
        let page = b
            .http
            .get_cookie(&format!("/alice/{repo}/settings/mirrors"), &session)
            .await;
        assert_eq!(page.status, 200, "GET mirrors of {repo}");
        assert!(
            !page.body.contains(&token),
            "the token leaked into the mirror page of {repo}"
        );
    }
    assert!(
        !server_log(&b).contains(&token),
        "the token leaked into the server log"
    );
}