"""安全扫描:主动构造攻击输入,而不是读代码猜。

四类:凭证泄漏、路径穿越、越权、存在性预言机。
"""
from __future__ import annotations

import uuid

import httpx
import pytest
import respx

from app.domains.github_org import constants as const
from app.domains.github_org import service as svc
from app.domains.github_org.models import GithubOrgJoinRequest
from app.domains.github_org.validators import normalize_github_login
from app.integrations.drivers.github_org import GithubOrgApiDriver, _redact

_API = "https://api.github.com"
# 只给需要 DB / 事件循环的用例挂 mark —— 纯函数用例是同步的。
_async = pytest.mark.asyncio


# ── 凭证泄漏 ─────────────────────────────────────────────────────


@_async
@respx.mock
async def test_no_credential_material_survives_an_error_path():
    """错误片段会落进 ``provisioning_tasks.error_message`` 与集成的
    ``last_error_message`` —— 两处管理员看得见。"""
    token = "ghp_" + "a" * 36
    pem = (
        "-----BEGIN RSA PRIVATE KEY-----\n"
        "MIIEowIBAAKCAQEAsecretbodyhere\n"
        "-----END RSA PRIVATE KEY-----"
    )
    driver = GithubOrgApiDriver(
        integration_id="x",
        config={"auth_mode": "pat", "token": token, "private_key": pem},
    )
    # GitHub 把请求原样回显在错误里(真实发生过的形态)。
    respx.mock.put(f"{_API}/orgs/o/memberships/u").mock(
        return_value=httpx.Response(
            400, text=f"bad request; Authorization: Bearer {token}; key={pem}",
        ),
    )
    res = await driver.invite_member("o", "u")
    blob = res.error_message or ""
    assert token not in blob
    assert "MIIEow" not in blob
    assert "BEGIN RSA PRIVATE KEY" not in blob


def test_redact_handles_every_github_token_prefix():
    for prefix in ("ghp_", "ghs_", "ghu_", "gho_", "ghr_", "github_pat_"):
        raw = f"leaked {prefix}{'x' * 30} here"
        assert f"{prefix}{'x' * 30}" not in _redact(raw)


# ── 路径穿越 ─────────────────────────────────────────────────────


def test_handle_validator_refuses_traversal_and_injection():
    """handle 曾经来自自助申请的表单。"""
    for bad in (
        "../admin", "a/../b", "a%2F..%2Fb", "a b", "a\nb", "-lead", "trail-",
        "a" * 40, "", "   ", "a--b",
    ):
        with pytest.raises(Exception):
            normalize_github_login(bad)


def test_url_shaped_handles_are_narrowed_not_trusted():
    """接受粘贴的主页链接,但只取第一段 —— 不能被带去别的路径。"""
    assert normalize_github_login("https://github.com/octocat/repo") == "octocat"
    assert normalize_github_login("github.com/octocat?x=1") == "octocat"


@_async
@respx.mock
async def test_org_slug_is_escaped_in_the_request_path():
    driver = GithubOrgApiDriver(
        integration_id="x", config={"auth_mode": "pat", "token": "ghp_" + "b" * 36},
    )
    # Matched on the REQUEST's ``raw_path``, not by handing respx a URL
    # pattern. respx compares a normalised URL, and normalisation is
    # exactly what this test exists to rule out: ``%2F`` and ``/`` compare
    # equal there, so a driver that had stopped escaping would still
    # match the pattern and the test would pass while the property was
    # gone. ``raw_path`` is the bytes httpx actually puts on the wire.
    route = respx.mock.delete(url__regex=r".*").mock(
        return_value=httpx.Response(204),
    )
    await driver.remove_member("../admin", "octocat")
    assert route.call_count == 1
    sent = route.calls[0].request
    assert sent.url.raw_path == b"/orgs/..%2Fadmin/memberships/octocat", (
        f"org slug reached the wire unescaped: {sent.url.raw_path!r} — a "
        "bare '/' here would let a slug walk out of the /orgs/ segment"
    )


# ── 越权 ─────────────────────────────────────────────────────────


async def _org(db):
    org = await svc.create_org(
        db, slug="pingcap-inc", display_name="X", feishu_option_value="opt",
    )
    await db.commit()
    return org


@_async
async def test_employee_cannot_reach_any_admin_github_endpoint(
    client, employee_headers, db_session,
):
    org = await _org(db_session)
    probes = [
        ("get", "/api/v1/github/orgs", None),
        ("get", "/api/v1/github/memberships", None),
        ("get", "/api/v1/github/auto-join-rules", None),
        ("post", "/api/v1/github/orgs", {"slug": "x", "display_name": "x"}),
        ("post", "/api/v1/github/auto-join-rules",
         {"org_id": org.id, "match_department_path": "E"}),
        ("post", f"/api/v1/github/orgs/{org.id}/reconcile", {}),
        ("post", "/api/v1/github/import-existing", None),
        ("delete", f"/api/v1/github/orgs/{org.id}", None),
    ]
    for method, path, body in probes:
        fn = getattr(client, method)
        resp = await (
            fn(path, headers=employee_headers, json=body)
            if body is not None else fn(path, headers=employee_headers)
        )
        assert resp.status_code == 403, f"{method} {path} → {resp.status_code}"


@_async
async def test_auditor_cannot_write_anything(
    client, auditor_headers, db_session,
):
    """审计能看见「谁能往我们的仓库里写代码」,但不替 org owner 决定。"""
    org = await _org(db_session)
    writes = [
        ("post", "/api/v1/github/orgs", {"slug": "y", "display_name": "y"}),
        ("patch", f"/api/v1/github/orgs/{org.id}", {"is_active": False}),
        ("delete", f"/api/v1/github/orgs/{org.id}", None),
        ("post", "/api/v1/github/import-existing", None),
    ]
    for method, path, body in writes:
        fn = getattr(client, method)
        resp = await (
            fn(path, headers=auditor_headers, json=body)
            if body is not None else fn(path, headers=auditor_headers)
        )
        assert resp.status_code == 403, f"{method} {path} → {resp.status_code}"


@_async
async def test_self_service_cannot_read_someone_elses_requests(
    client, employee_headers, admin_user, db_session,
):
    other, _ = admin_user
    org = await _org(db_session)
    row = GithubOrgJoinRequest(
        id=str(uuid.uuid4()), person_id=str(other.id), org_id=org.id,
        github_login="theirs", status=const.JOIN_REQUEST_STATUS_PENDING,
    )
    db_session.add(row)
    await db_session.commit()

    resp = await client.get("/api/v1/me/github/requests", headers=employee_headers)
    assert resp.status_code == 200
    assert all(r["id"] != row.id for r in resp.json()["data"])


@_async
async def test_self_service_cannot_read_someone_elses_memberships(
    client, employee_headers, admin_user, db_session,
):
    other, _ = admin_user
    org = await _org(db_session)
    await svc.upsert_membership(
        db_session, org=org, github_login="theirs", person_id=str(other.id),
        state=const.MEMBERSHIP_STATE_ACTIVE, source=const.SOURCE_IMPORT,
    )
    await db_session.commit()
    resp = await client.get(
        "/api/v1/me/github/memberships", headers=employee_headers,
    )
    assert resp.json()["data"] == []


# ── 存在性预言机 ─────────────────────────────────────────────────


@_async
async def test_cancelling_a_foreign_request_is_indistinguishable_from_missing(
    client, employee_headers, admin_user, db_session,
):
    """撤销别人的单子与撤销一张不存在的单子,回答必须一模一样 ——
    否则这是一个可以枚举「哪些 id 是真的」的口子。"""
    other, _ = admin_user
    org = await _org(db_session)
    row = GithubOrgJoinRequest(
        id=str(uuid.uuid4()), person_id=str(other.id), org_id=org.id,
        github_login="theirs", status=const.JOIN_REQUEST_STATUS_PENDING,
    )
    db_session.add(row)
    await db_session.commit()

    foreign = await client.post(
        f"/api/v1/me/github/requests/{row.id}/cancel", headers=employee_headers,
    )
    missing = await client.post(
        f"/api/v1/me/github/requests/{uuid.uuid4()}/cancel",
        headers=employee_headers,
    )
    assert foreign.status_code == missing.status_code == 404
    assert (
        foreign.json()["errors"][0]["code"]
        == missing.json()["errors"][0]["code"]
    )