"""Gate 3.4: Core verifies app access tokens fail-closed, and checks the Cloud session live."""

import time

import httpx
import pytest
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric import rsa
from jose import jwt

from src.domain.project_auth.app_token import AppTokenError, verify_app_access_token
from src.integrations.cloud_session_introspection import (
    SessionCheckUnavailable,
    mint_introspection_capability,
    session_is_active,
)

PROJECT = "192dd090-3cdb-4842-a82e-51dbd0414527"
SECRET = "x" * 40


def _keypair():
    key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
    private = key.private_bytes(
        serialization.Encoding.PEM, serialization.PrivateFormat.PKCS8, serialization.NoEncryption()
    ).decode()
    public = key.public_key().public_bytes(
        serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo
    ).decode()
    return private, public


PRIVATE, PUBLIC = _keypair()
OTHER_PRIVATE, _ = _keypair()


def _claims(**overrides):
    now = int(time.time())
    claims = {
        "iss": "p2hs-cloud",
        "aud": ["p2hs-cloud-app", "p2hs-core-authz"],
        "token_use": "access",
        "sub": "usr_owner",
        "sid": "sess_1",
        "project_id": PROJECT,
        "iat": now,
        "exp": now + 900,
    }
    claims.update(overrides)
    return {k: v for k, v in claims.items() if v is not None}


def _token(claims=None, *, private=PRIVATE, kid="k1", alg="RS256"):
    return jwt.encode(claims or _claims(), private, algorithm=alg, headers={"kid": kid})


KEYS = {"k1": PUBLIC}


def test_valid_token_resolves_the_actor():
    actor = verify_app_access_token(_token(), project_id=PROJECT, keys=KEYS)
    assert (actor.cloud_app_user_id, actor.session_id, actor.project_id) == ("usr_owner", "sess_1", PROJECT)


@pytest.mark.parametrize(
    "token, code",
    [
        (_token(alg="HS256", private="s" * 40), "bad_alg"),
        (_token(kid="unknown"), "unknown_kid"),
        (_token(_claims(iss="someone-else")), "bad_claims"),
        (_token(_claims(aud=["p2hs-cloud-app"])), "bad_claims"),
        (_token(_claims(token_use="refresh")), "bad_token_use"),
        (_token(_claims(project_id="00000000-0000-0000-0000-000000000000")), "project_mismatch"),
        (_token(_claims(exp=int(time.time()) - 3600, iat=int(time.time()) - 7200)), "expired"),
        (_token(_claims(sid=None)), "missing_subject_or_session"),
        (_token(_claims(sub="")), "missing_subject_or_session"),
        (_token(_claims(iat=int(time.time()) + 3600)), "iat_in_future"),
        (_token(private=OTHER_PRIVATE), "bad_signature"),
        ("not-a-jwt", "malformed"),
    ],
)
def test_every_failure_is_rejected_with_a_stable_code(token, code):
    with pytest.raises(AppTokenError) as raised:
        verify_app_access_token(token, project_id=PROJECT, keys=KEYS)
    assert raised.value.code == code


def test_revoked_or_absent_key_cannot_verify():
    # The caller passes only usable keys; a token signed by a key not in the map is unknown.
    with pytest.raises(AppTokenError) as raised:
        verify_app_access_token(_token(), project_id=PROJECT, keys={})
    assert raised.value.code == "unknown_kid"


def _introspection_transport(status_code: int, body: dict | None = None, *, fail: bool = False):
    def handler(request: httpx.Request) -> httpx.Response:
        if fail:
            raise httpx.ConnectError("boom", request=request)
        assert request.url.path == f"/v1/cloud/core/projects/{PROJECT}/auth/sessions/introspect"
        assert request.headers["Authorization"].startswith("Bearer ")
        return httpx.Response(status_code, json=body)

    return httpx.MockTransport(handler)


@pytest.mark.asyncio
async def test_active_session_is_allowed():
    transport = _introspection_transport(200, {"active": True})
    assert await session_is_active(
        "https://cloud.example", SECRET, PROJECT,
        cloud_app_user_id="usr_owner", session_id="sess_1", transport=transport,
    ) is True


@pytest.mark.asyncio
async def test_inactive_session_is_denied():
    transport = _introspection_transport(200, {"active": False})
    assert await session_is_active(
        "https://cloud.example", SECRET, PROJECT,
        cloud_app_user_id="usr_owner", session_id="sess_1", transport=transport,
    ) is False


@pytest.mark.asyncio
async def test_unexpected_answer_is_denied_not_allowed():
    transport = _introspection_transport(200, {"active": "yes"})
    assert await session_is_active(
        "https://cloud.example", SECRET, PROJECT,
        cloud_app_user_id="usr_owner", session_id="sess_1", transport=transport,
    ) is False


@pytest.mark.asyncio
async def test_cloud_errors_raise_instead_of_denying_as_revoked():
    for transport in (_introspection_transport(401, {"error": "unauthorized"}),
                      _introspection_transport(0, fail=True)):
        with pytest.raises(SessionCheckUnavailable):
            await session_is_active(
                "https://cloud.example", SECRET, PROJECT,
                cloud_app_user_id="usr_owner", session_id="sess_1", transport=transport,
            )


def test_introspection_capability_is_narrow_and_short_lived():
    token = mint_introspection_capability(PROJECT, SECRET, now=1_000_000)
    claims = jwt.decode(token, SECRET, algorithms=["HS256"], audience="p2hs-cloud-auth-projection", options={"verify_exp": False})
    assert claims["capability"] == "auth_session.introspect"
    assert claims["project_id"] == PROJECT
    assert claims["exp"] - claims["iat"] <= 300