# -*- coding: utf-8 -*-
# Copyright (c) 2025 ContextForge Contributors.
# SPDX-License-Identifier: Apache-2.0

"""Location: ./tests/playwright/security/test_api_abuse_hardening.py
Copyright 2026
SPDX-License-Identifier: Apache-2.0
Authors: Mihai Criveti

API abuse hardening tests: mass assignment, BOLA, HPP, XSS, and path traversal.
"""

# Future
from __future__ import annotations

# Standard
from contextlib import suppress
from typing import Any
import uuid

# Third-Party
from playwright.sync_api import APIRequestContext, Playwright
import pytest

# First-Party
from mcpgateway.utils.create_jwt_token import _create_jwt_token

# Local
from .conftest import BASE_URL, TEST_PASSWORD

_UNSET = object()


def _extract_token_id(response_json: dict[str, Any]) -> str | None:
    token_obj = response_json.get("token", response_json)
    return token_obj.get("id") or token_obj.get("token_id")


def _extract_servers(response_json: Any) -> list[dict[str, Any]]:
    if isinstance(response_json, list):
        return response_json
    if isinstance(response_json, dict):
        maybe_servers = response_json.get("servers")
        if isinstance(maybe_servers, list):
            return maybe_servers
    return []


def _make_jwt(email: str, *, is_admin: bool, teams: object = _UNSET) -> str:
    payload: dict[str, Any] = {"sub": email}
    if teams is not _UNSET:
        payload["teams"] = teams
    return _create_jwt_token(
        payload,
        user_data={"email": email, "is_admin": is_admin, "auth_provider": "local"},
    )


def _api_context(playwright: Playwright, token: str) -> APIRequestContext:
    return playwright.request.new_context(
        base_url=BASE_URL,
        extra_http_headers={"Authorization": f"Bearer {token}", "Accept": "application/json"},
    )


@pytest.fixture
def hpp_servers(playwright: Playwright):
    admin_ctx = _api_context(playwright, _make_jwt("admin@example.com", is_admin=True, teams=None))
    team_id: str | None = None
    public_server_id: str | None = None
    team_server_id: str | None = None
    try:
        team_resp = admin_ctx.post(
            "/teams/",
            data={"name": f"hpp-team-{uuid.uuid4().hex[:8]}", "description": "HPP test team", "visibility": "private"},
        )
        assert team_resp.status in (200, 201), f"Failed creating team for HPP test: {team_resp.status} {team_resp.text()}"
        team_id = team_resp.json()["id"]

        public_server_resp = admin_ctx.post(
            "/servers",
            data={"server": {"name": f"hpp-public-{uuid.uuid4().hex[:8]}"}, "team_id": None, "visibility": "public"},
        )
        team_server_resp = admin_ctx.post(
            "/servers",
            data={"server": {"name": f"hpp-team-{uuid.uuid4().hex[:8]}"}, "team_id": team_id, "visibility": "team"},
        )
        assert public_server_resp.status in (200, 201), f"Failed creating public server: {public_server_resp.status} {public_server_resp.text()}"
        assert team_server_resp.status in (200, 201), f"Failed creating team server: {team_server_resp.status} {team_server_resp.text()}"

        public_server_id = public_server_resp.json()["id"]
        team_server_id = team_server_resp.json()["id"]

        yield {"team_id": team_id, "public_server_id": public_server_id, "team_server_id": team_server_id}
    finally:
        if team_server_id:
            with suppress(Exception):
                admin_ctx.delete(f"/servers/{team_server_id}")
        if public_server_id:
            with suppress(Exception):
                admin_ctx.delete(f"/servers/{public_server_id}")
        if team_id:
            with suppress(Exception):
                admin_ctx.delete(f"/teams/{team_id}")
        admin_ctx.dispose()


class TestAPISecurityAbuseCases:
    """Coverage for high-priority API abuse vectors from manual security plans."""

    def test_mass_assignment_cannot_override_token_owner(self, admin_api: APIRequestContext):
        token_name = f"mass-assign-{uuid.uuid4().hex[:8]}"
        response = admin_api.post(
            "/tokens",
            data={
                "name": token_name,
                "expires_in_days": 1,
                "user_email": "victim@example.com",
                "is_admin": True,
            },
        )

        token_id: str | None = None
        if response.status in (200, 201):
            body = response.json()
            token_id = _extract_token_id(body)
            token_obj = body.get("token", body)
            assert token_obj.get("user_email") == "admin@example.com"
        else:
            assert response.status in (400, 422), f"Unexpected response for mass-assignment probe: {response.status} {response.text()}"

        if token_id:
            with suppress(Exception):
                admin_api.delete(f"/tokens/{token_id}")

    def test_bola_user_cannot_read_or_revoke_another_users_token(self, admin_api: APIRequestContext, playwright: Playwright):
        email_a = f"bola-a-{uuid.uuid4().hex[:8]}@example.com"
        email_b = f"bola-b-{uuid.uuid4().hex[:8]}@example.com"

        created_token_id: str | None = None
        ctx_a = None
        ctx_b = None
        try:
            for email in (email_a, email_b):
                create_user_resp = admin_api.post(
                    "/auth/email/admin/users",
                    data={"email": email, "password": TEST_PASSWORD, "full_name": "BOLA Test User"},
                )
                assert create_user_resp.status in (200, 201), f"Failed creating {email}: {create_user_resp.status} {create_user_resp.text()}"

            ctx_a = _api_context(playwright, _make_jwt(email_a, is_admin=False, teams=[]))
            ctx_b = _api_context(playwright, _make_jwt(email_b, is_admin=False, teams=[]))

            create_token_resp = ctx_a.post(
                "/tokens",
                data={"name": f"bola-token-{uuid.uuid4().hex[:8]}", "expires_in_days": 1},
            )
            assert create_token_resp.status in (200, 201), f"User A failed to create token: {create_token_resp.status} {create_token_resp.text()}"
            created_token_id = _extract_token_id(create_token_resp.json())
            assert created_token_id

            read_other_resp = ctx_b.get(f"/tokens/{created_token_id}")
            revoke_other_resp = ctx_b.delete(f"/tokens/{created_token_id}")

            assert read_other_resp.status in (403, 404), f"User B should not read User A token: {read_other_resp.status} {read_other_resp.text()}"
            assert revoke_other_resp.status in (403, 404), f"User B should not revoke User A token: {revoke_other_resp.status} {revoke_other_resp.text()}"
        finally:
            if ctx_a:
                if created_token_id:
                    with suppress(Exception):
                        ctx_a.delete(f"/tokens/{created_token_id}")
                ctx_a.dispose()
            if ctx_b:
                ctx_b.dispose()
            with suppress(Exception):
                admin_api.delete(f"/auth/email/admin/users/{email_a}")
            with suppress(Exception):
                admin_api.delete(f"/auth/email/admin/users/{email_b}")

    def test_query_parameter_pollution_does_not_bypass_public_only_scope(self, hpp_servers: dict[str, str], playwright: Playwright):
        public_only_token = _make_jwt("admin@example.com", is_admin=True)
        ctx = _api_context(playwright, public_only_token)
        try:
            response = ctx.get("/servers?visibility=public&visibility=team")
            status_code = response.status
            payload = response.json()
        finally:
            ctx.dispose()

        assert status_code == 200, f"Unexpected status for HPP probe: {status_code}"
        ids = {server["id"] for server in _extract_servers(payload)}
        assert hpp_servers["team_server_id"] not in ids, "Public-only token must never obtain team-scoped resources via duplicate query params"

    def test_xss_payload_in_server_name_is_rejected_or_sanitized(self, admin_api: APIRequestContext):
        response = admin_api.post(
            "/servers",
            data={
                "server": {"name": '<script>alert("xss")</script>', "description": "xss probe"},
                "team_id": None,
                "visibility": "public",
            },
        )

        created_server_id: str | None = None
        if response.status in (200, 201):
            body = response.json()
            created_server_id = body.get("id")
            assert "<script>" not in (body.get("name", "")).lower()
        else:
            assert response.status in (400, 422), f"Unexpected XSS probe response: {response.status} {response.text()}"

        if created_server_id:
            with suppress(Exception):
                admin_api.delete(f"/servers/{created_server_id}")

    def test_path_traversal_payload_in_resource_uri_is_rejected(self, admin_api: APIRequestContext):
        response = admin_api.post(
            "/resources",
            data={
                "resource": {"uri": "../../etc/passwd", "name": f"traversal-{uuid.uuid4().hex[:8]}", "content": "probe"},
                "team_id": None,
                "visibility": "private",
            },
        )

        if response.status in (200, 201):
            body = response.json()
            assert ".." not in str(body.get("uri", "")), "Path traversal sequences should not persist in stored URI"
            created_id = body.get("id")
            if created_id:
                with suppress(Exception):
                    admin_api.delete(f"/resources/{created_id}")
            pytest.fail("Traversal payload unexpectedly accepted without normalization/rejection")

        assert response.status in (400, 422), f"Traversal payload should be rejected, got {response.status}: {response.text()}"
