"""The encrypted OAuth token store: permissions, encryption, key handling and ownership (SEC-13)."""

from __future__ import annotations

import os
import stat
from pathlib import Path
from typing import Any

import pytest
from cryptography.fernet import Fernet
from pytest_httpx import HTTPXMock

from aegis_gitea_mcp import transports
from aegis_gitea_mcp.oauth.store import (
    STORE_FILENAME,
    OAuthStore,
    OAuthStoreError,
    SessionRecord,
    derive_fernet_key,
    token_hash,
)
from aegis_gitea_mcp.transports import run_http
from tests.test_oauth_helpers import (
    APP_CLIENT_SECRET,
    ENCRYPTION_KEY,
    USER_REFRESH,
    USER_TOKEN,
    browser,
    build_oauth_app,
    oauth_settings,
    sign_in,
)

OTHER_KEY = "another-key-0123456789abcdef0123456789"


@pytest.fixture(autouse=True)
def _env(mock_env: None) -> None:
    """The audit logger reads the process-wide settings."""


def a_session() -> SessionRecord:
    return SessionRecord(
        session_id="s1",
        client_id="c1",
        login="alice",
        is_admin=False,
        gitea_access_token=USER_TOKEN,
        gitea_refresh_token=USER_REFRESH,
        gitea_expires_at=None,
        scopes=["gitea"],
        resource="https://mcp.example.com/mcp",
        created_at=1e12,
    )


async def test_file_is_0600_and_a_new_directory_is_0700(tmp_path: Path) -> None:
    previous = os.umask(0)
    try:
        directory = tmp_path / "new" / "state"
        store = OAuthStore.load(directory, ENCRYPTION_KEY)
        store.sessions["s1"] = a_session()
        await store.save()
    finally:
        os.umask(previous)

    assert stat.S_IMODE(directory.stat().st_mode) == 0o700
    assert stat.S_IMODE((directory / STORE_FILENAME).stat().st_mode) == 0o600
    assert [entry.name for entry in directory.iterdir()] == [STORE_FILENAME]


def test_a_symlinked_store_is_refused_and_its_target_untouched(tmp_path: Path) -> None:
    target = tmp_path / "elsewhere"
    target.write_bytes(b"keep me")
    (tmp_path / STORE_FILENAME).symlink_to(target)

    with pytest.raises(OAuthStoreError, match="symlink"):
        OAuthStore.load(tmp_path, ENCRYPTION_KEY)
    assert target.read_bytes() == b"keep me"


async def test_a_symlink_swapped_in_later_is_not_written_through(tmp_path: Path) -> None:
    store = OAuthStore.load(tmp_path, ENCRYPTION_KEY)
    target = tmp_path / "elsewhere"
    target.write_bytes(b"keep me")
    (tmp_path / STORE_FILENAME).symlink_to(target)

    with pytest.raises(OAuthStoreError, match="symlink"):
        await store.save()
    assert target.read_bytes() == b"keep me"


async def test_the_file_holds_no_plaintext_and_our_tokens_only_as_hashes(
    tmp_path: Path, httpx_mock: HTTPXMock
) -> None:
    oauth = build_oauth_app(oauth_settings(data_dir=tmp_path))
    flow = sign_in(browser(oauth.app), httpx_mock, login="alice-the-user")

    raw = (tmp_path / STORE_FILENAME).read_bytes()

    for secret in (
        USER_TOKEN,
        USER_REFRESH,
        flow.client_secret,
        flow.tokens["access_token"],
        flow.tokens["refresh_token"],
        "alice-the-user",
        APP_CLIENT_SECRET,
        ENCRYPTION_KEY,
    ):
        assert secret.encode() not in raw
    plain = Fernet(derive_fernet_key(ENCRYPTION_KEY)).decrypt(raw).decode()
    assert flow.tokens["access_token"] not in plain
    assert flow.tokens["refresh_token"] not in plain
    assert token_hash(flow.tokens["access_token"]) in plain
    assert token_hash(flow.tokens["refresh_token"]) in plain


async def test_a_wrong_key_stops_startup_without_wiping_the_store(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
) -> None:
    store = OAuthStore.load(tmp_path, ENCRYPTION_KEY)
    store.sessions["s1"] = a_session()
    await store.save()
    before = (tmp_path / STORE_FILENAME).read_bytes()
    served: list[Any] = []
    monkeypatch.setattr(transports.uvicorn, "run", lambda *a, **k: served.append(a))

    code = run_http(oauth_settings(data_dir=tmp_path, token_encryption_key=OTHER_KEY))

    assert code == 2 and served == []
    err = capsys.readouterr().err
    assert "TOKEN_ENCRYPTION_KEY" in err and "move the file away" in err
    assert OTHER_KEY not in err and ENCRYPTION_KEY not in err
    assert (tmp_path / STORE_FILENAME).read_bytes() == before


def test_a_garbage_store_stops_startup_and_stays_on_disk(tmp_path: Path) -> None:
    (tmp_path / STORE_FILENAME).write_bytes(b"not a fernet token")

    with pytest.raises(OAuthStoreError):
        OAuthStore.load(tmp_path, ENCRYPTION_KEY)
    assert (tmp_path / STORE_FILENAME).read_bytes() == b"not a fernet token"


# SEC-13: who owns the directory


def fake_owner(monkeypatch: pytest.MonkeyPatch, directory: Path) -> None:
    """Report ``directory`` as owned by a different uid than the running process."""
    real_stat = os.stat

    def stat_other(path: Any, *args: Any, **kwargs: Any) -> os.stat_result:
        info = real_stat(path, *args, **kwargs)
        if Path(path) == directory:
            fields = list(info)
            fields[stat.ST_UID] = info.st_uid + 1
            return os.stat_result(fields)
        return info

    monkeypatch.setattr(os, "stat", stat_other)


async def test_a_directory_owned_by_another_user_is_refused_on_load_and_save(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    directory = tmp_path / "state"
    directory.mkdir(mode=0o700)
    store = OAuthStore(directory, ENCRYPTION_KEY)
    fake_owner(monkeypatch, directory)

    with pytest.raises(OAuthStoreError, match="owned by another user"):
        OAuthStore.load(directory, ENCRYPTION_KEY)
    with pytest.raises(OAuthStoreError, match="owned by another user"):
        await store.save()
    assert list(directory.iterdir()) == []


async def test_a_group_or_world_writable_directory_that_cannot_be_tightened_is_refused(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    directory = tmp_path / "state"
    directory.mkdir()
    directory.chmod(0o777)

    def refuse(self: Path, mode: int) -> None:
        raise PermissionError("not permitted")

    monkeypatch.setattr(Path, "chmod", refuse)

    with pytest.raises(OAuthStoreError, match="writable by group or others"):
        OAuthStore.load(directory, ENCRYPTION_KEY)
    with pytest.raises(OAuthStoreError, match="writable by group or others"):
        await OAuthStore(directory, ENCRYPTION_KEY).save()
    assert list(directory.iterdir()) == []


async def test_a_loose_directory_we_own_is_tightened_not_refused(tmp_path: Path) -> None:
    directory = tmp_path / "state"
    directory.mkdir()
    directory.chmod(0o777)

    await OAuthStore.load(directory, ENCRYPTION_KEY).save()

    assert stat.S_IMODE(directory.stat().st_mode) == 0o700


def test_startup_exits_2_for_a_directory_owned_by_someone_else(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
) -> None:
    directory = tmp_path / "state"
    directory.mkdir(mode=0o700)
    fake_owner(monkeypatch, directory)
    served: list[Any] = []
    monkeypatch.setattr(transports.uvicorn, "run", lambda *a, **k: served.append(a))

    assert run_http(oauth_settings(data_dir=directory)) == 2

    assert served == []
    assert "owned by another user" in capsys.readouterr().err