from __future__ import annotations

import base64
import hashlib
from datetime import UTC, datetime, timedelta

import jwt
import pytest
from cryptography.hazmat.primitives.asymmetric import rsa

NOW = datetime(2026, 8, 2, 8, 0, tzinfo=UTC)


def _b64int(value: int) -> str:
    size = (value.bit_length() + 7) // 8
    return base64.urlsafe_b64encode(value.to_bytes(size, "big")).rstrip(b"=").decode()


def _key_material():
    private = rsa.generate_private_key(public_exponent=65537, key_size=2048)
    numbers = private.public_key().public_numbers()
    jwk = {"kty": "RSA", "kid": "test-key", "alg": "RS256", "use": "sig", "n": _b64int(numbers.n), "e": _b64int(numbers.e)}
    return private, jwk


def test_idp_config_rejects_unsafe_urls_redirects_and_secret_values(monkeypatch):
    from app.core.system.enterprise_identity import IdentityPolicyError, IdpConfig

    valid = {
        "provider_uid": "idp-1", "version": 1, "issuer": "https://login.example.com/tenant",
        "client_id": "dataops", "secret_ref": "env:DATAOPS_OIDC_CLIENT_SECRET",
        "authorization_endpoint": "https://login.example.com/oauth2/authorize",
        "token_endpoint": "https://login.example.com/oauth2/token",
        "jwks_uri": "https://login.example.com/.well-known/jwks.json",
        "redirect_uris": ("https://dataops.example.com/api/system/identity/callback",),
        "post_login_redirect_uris": ("https://dataops.example.com/login/callback",),
        "algorithms": ("RS256",), "mapping_version": "mapping-v1",
    }
    config = IdpConfig(**valid)
    config.validate()
    monkeypatch.setenv("DATAOPS_OIDC_CLIENT_SECRET", "not-returned")
    assert config.resolve_secret() == "not-returned"
    assert "not-returned" not in repr(config)

    for update in (
        {"issuer": "http://login.example.com"},
        {"issuer": "https://127.0.0.1"},
        {"issuer": "https://localhost"},
        {"secret_ref": "plain-secret"},
        {"secret_ref": "env:AWS_SECRET_ACCESS_KEY"},
        {"secret_ref": "env:OIDC_TEST_SECRET"},
        {"algorithms": ("none",)},
        {"redirect_uris": ("https://dataops.example.com/callback#token",)},
        {"post_login_redirect_uris": ("javascript:alert(1)",)},
        {"post_login_redirect_uris": ("https://localhost/login/callback",)},
    ):
        payload = {**valid, **update}
        with pytest.raises(IdentityPolicyError):
            IdpConfig(**payload).validate()


def test_oidc_pkce_state_nonce_signature_and_replay_are_fail_closed():
    from app.core.system.enterprise_identity import IdentityPolicyError, IdpConfig
    from app.core.system.identity_repository import MemoryIdentityRepository
    from app.core.system.oidc import OidcClient

    config = IdpConfig(
        provider_uid="idp-1", version=1, issuer="https://login.example.com/tenant",
        client_id="dataops", secret_ref="env:DATAOPS_OIDC_TEST_SECRET",
        authorization_endpoint="https://login.example.com/authorize",
        token_endpoint="https://login.example.com/token",
        jwks_uri="https://login.example.com/jwks", redirect_uris=("https://dataops.example.com/callback",),
        post_login_redirect_uris=("https://dataops.example.com/login/callback",),
        algorithms=("RS256",), mapping_version="mapping-v1",
    )
    private, jwk = _key_material()
    repo = MemoryIdentityRepository()
    client = OidcClient(repo, clock=lambda: NOW)
    started = client.begin(config, "https://dataops.example.com/callback")
    assert started.code_challenge_method == "S256"
    assert started.code_challenge == base64.urlsafe_b64encode(
        hashlib.sha256(started.code_verifier.encode()).digest()
    ).rstrip(b"=").decode()

    claims = {"iss": config.issuer, "aud": config.client_id, "sub": "employee-1", "nonce": started.nonce,
              "iat": int(NOW.timestamp()), "exp": int((NOW + timedelta(minutes=5)).timestamp()),
              "preferred_username": "alice", "name": "Alice", "department": "ops", "groups": ["stewards"]}
    token = jwt.encode(claims, private, algorithm="RS256", headers={"kid": "test-key"})
    verified = client.verify_callback(config, state=started.state, redirect_uri="https://dataops.example.com/callback",
                                      code_verifier=started.code_verifier, id_token=token, jwks={"keys": [jwk]})
    assert verified["sub"] == "employee-1"
    with pytest.raises(IdentityPolicyError, match="state"):
        client.verify_callback(config, state=started.state, redirect_uri="https://dataops.example.com/callback",
                               code_verifier=started.code_verifier, id_token=token, jwks={"keys": [jwk]})

    forged = jwt.encode({**claims, "nonce": "wrong"}, "symmetric", algorithm="HS256", headers={"kid": "test-key"})
    second = client.begin(config, "https://dataops.example.com/callback")
    with pytest.raises(IdentityPolicyError):
        client.verify_callback(config, state=second.state, redirect_uri="https://dataops.example.com/callback",
                               code_verifier=second.code_verifier, id_token=forged, jwks={"keys": [jwk]})


def test_oidc_transport_rejects_private_dns_redirects_oversize_and_bad_json():
    from app.core.system.enterprise_identity import (
        IdentityPolicyError,
        IdentityUpstreamError,
        IdpConfig,
    )
    from app.core.system.oidc import OidcTransport

    config = IdpConfig(
        provider_uid="idp-1", version=1, issuer="https://login.example.com/tenant",
        client_id="dataops", secret_ref="env:DATAOPS_OIDC_TEST_SECRET",
        authorization_endpoint="https://login.example.com/authorize",
        token_endpoint="https://login.example.com/token", jwks_uri="https://login.example.com/jwks",
        redirect_uris=("https://dataops.example.com/callback",),
        post_login_redirect_uris=("https://dataops.example.com/login/callback",),
        algorithms=("RS256",), mapping_version="mapping-v1",
    )

    class Response:
        def __init__(self, status, body, content_type="application/json"):
            self.status_code = status
            self.body = body
            self.headers = {"Content-Type": content_type, "Content-Length": str(len(body))}

        def iter_content(self, chunk_size):
            del chunk_size
            yield self.body

        def close(self):
            return None

    class Http:
        def __init__(self, response):
            self.response = response

        def get(self, url, **kwargs):
            del url, kwargs
            return self.response

    def public_dns(*args, **kwargs):
        del args, kwargs
        return [(2, 1, 6, "", ("93.184.216.34", 443))]

    def private_dns(*args, **kwargs):
        del args, kwargs
        return [(2, 1, 6, "", ("10.0.0.8", 443))]
    with pytest.raises(IdentityPolicyError, match="non-public"):
        OidcTransport(http=Http(Response(200, b'{"keys":[]}')), resolver=private_dns).fetch_jwks(config)
    with pytest.raises(IdentityUpstreamError, match="redirect"):
        OidcTransport(http=Http(Response(302, b"{}")), resolver=public_dns).fetch_jwks(config)
    with pytest.raises(IdentityUpstreamError, match="too large"):
        OidcTransport(http=Http(Response(200, b"x" * 33)), resolver=public_dns,
                      max_response_bytes=32).fetch_jwks(config)
    with pytest.raises(IdentityUpstreamError, match="invalid JSON"):
        OidcTransport(http=Http(Response(200, b"not-json")), resolver=public_dns).fetch_jwks(config)
    with pytest.raises(IdentityUpstreamError, match="must be JSON"):
        OidcTransport(http=Http(Response(200, b"{}", "text/plain")), resolver=public_dns).fetch_jwks(config)


def test_oidc_production_transport_pins_validated_ip_and_preserves_tls_hostname():
    from app.core.system.enterprise_identity import IdpConfig
    from app.core.system.oidc import OidcTransport

    config = IdpConfig(
        provider_uid="idp-1", version=1, issuer="https://login.example.com/tenant",
        client_id="dataops", secret_ref="env:DATAOPS_OIDC_TEST_SECRET",
        authorization_endpoint="https://login.example.com/authorize",
        token_endpoint="https://login.example.com/token", jwks_uri="https://login.example.com/jwks?tenant=one",
        redirect_uris=("https://dataops.example.com/callback",),
        post_login_redirect_uris=("https://dataops.example.com/login/callback",),
        algorithms=("RS256",), mapping_version="mapping-v1",
    )
    observed = {"resolver_calls": 0, "pool": [], "request": []}

    def resolver(host, port, **kwargs):
        observed["resolver_calls"] += 1
        assert host == "login.example.com" and port == 443
        assert kwargs["type"]
        return [(2, 1, 6, "", ("93.184.216.34", 443))]

    class Response:
        status = 200
        headers = {"Content-Type": "application/json", "Content-Length": "11"}

        def stream(self, **kwargs):
            assert kwargs == {"amt": 64 * 1024, "decode_content": True}
            yield b'{"keys":[]}'

        def release_conn(self):
            return None

    class Pool:
        def __init__(self, **kwargs):
            observed["pool"].append(kwargs)

        def urlopen(self, method, target, **kwargs):
            observed["request"].append((method, target, kwargs))
            return Response()

        def close(self):
            return None

    payload = OidcTransport(resolver=resolver, pool_factory=Pool).fetch_jwks(config)
    assert payload == {"keys": []}
    assert observed["resolver_calls"] == 1
    pool = observed["pool"][0]
    assert pool["host"] == "93.184.216.34"
    assert pool["server_hostname"] == "login.example.com"
    assert pool["assert_hostname"] == "login.example.com"
    assert pool["cert_reqs"] == "CERT_REQUIRED" and pool["ca_certs"]
    method, target, request_options = observed["request"][0]
    assert method == "GET" and target == "/jwks?tenant=one"
    assert request_options["headers"]["Host"] == "login.example.com"
    assert request_options["redirect"] is False and request_options["retries"] is False


def test_claim_mapping_is_deterministic_versioned_and_default_deny():
    from app.core.system.enterprise_identity import ClaimMapper, IdentityPolicyError

    mapper = ClaimMapper(
        version="mapping-v7",
        group_rules={"stewards": {"roles": ["editor"], "business_domain_uids": ["domain-b", "domain-a"],
                                    "object_types": ["dataset"], "environments": ["prod"],
                                    "data_scopes": ["plant-1"]}},
    )
    claims = {"sub": "s-1", "preferred_username": "alice", "name": "Alice", "department": "ops",
              "groups": ["stewards"]}
    result = mapper.map(claims)
    assert result.roles == ("editor",)
    assert result.business_domain_uids == ("domain-a", "domain-b")
    assert result.mapping_version == "mapping-v7"
    assert len(result.claims_digest) == 64
    assert "Alice" not in repr(result.evidence)
    with pytest.raises(IdentityPolicyError):
        mapper.map({**claims, "groups": ["unknown"]})
    for malformed_claims in (
        {**claims, "groups": "stewards"},
        {**claims, "groups": ["stewards", 1]},
        {**claims, "groups": []},
        {**claims, "sub": {"value": "s-1"}},
        {**claims, "preferred_username": ["alice"]},
        {**claims, "name": ""},
        {**claims, "department": "x" * 301},
    ):
        with pytest.raises(IdentityPolicyError):
            mapper.map(malformed_claims)
    with pytest.raises(IdentityPolicyError):
        ClaimMapper(version="x", group_rules={"bad": {"roles": ["superuser"]}})
    for version, rules in (
        ("", {"group": {"roles": ["viewer"]}}),
        ("v1", {}),
        ("v1", {"group": "viewer"}),
        ("v1", {"group": {"roles": "viewer"}}),
        ("v1", {"group": {"roles": ["viewer"], "unknown_scope": ["x"]}}),
        ("v1", {"group": {"roles": ["viewer"], "data_scopes": [1]}}),
    ):
        with pytest.raises(IdentityPolicyError):
            ClaimMapper(version=version, group_rules=rules)


def test_directory_events_are_idempotent_conflict_safe_and_leaver_revokes():
    from app.core.system.enterprise_identity import (
        DirectorySynchronizer,
        IdentityPolicyError,
    )
    from app.core.system.identity_repository import MemoryIdentityRepository
    from app.core.system.identity_sessions import SessionManager

    repo = MemoryIdentityRepository()
    sessions = SessionManager(repo, secret="a-long-test-secret", clock=lambda: NOW)
    sync = DirectorySynchronizer(repo, sessions, clock=lambda: NOW)
    first = sync.apply(provider_uid="idp-a", source="scim", source_event_id="e1", cursor="10", cursor_sequence=10, event_type="JOINER",
                       subject="s1", attributes={"username": "alice", "roles": ["viewer"]})
    sessions.create(provider_uid="idp-a", subject="s1", user_uid=first["user_uid"], roles=["viewer"], identity_source="oidc")
    sessions.create(provider_uid="idp-b", subject="s1", user_uid="other", roles=["viewer"], identity_source="oidc")
    assert sync.apply(provider_uid="idp-a", source="scim", source_event_id="e1", cursor="10", cursor_sequence=10, event_type="JOINER",
                      subject="s1", attributes={"username": "alice", "roles": ["viewer"]})["idempotent"] is True
    with pytest.raises(IdentityPolicyError):
        sync.apply(provider_uid="idp-a", source="scim", source_event_id="e1", cursor="10", cursor_sequence=10, event_type="MOVER",
                   subject="s1", attributes={"username": "alice", "roles": ["editor"]})
    with pytest.raises(IdentityPolicyError, match="monotonic"):
        sync.apply(provider_uid="idp-a", source="scim", source_event_id="old", cursor="9", cursor_sequence=9,
                   event_type="MOVER", subject="s1", attributes={"username": "alice", "roles": ["viewer"]})
    assert len(repo.directory_conflicts) == 2
    sync.apply(provider_uid="idp-a", source="scim", source_event_id="e2", cursor="11", cursor_sequence=11, event_type="LEAVER",
               subject="s1", attributes={"username": "alice", "roles": []})
    assert all(item["status"] == "revoked" for item in repo.list_sessions("idp-a", "s1"))
    assert repo.list_sessions("idp-b", "s1")[0]["status"] == "active"


def test_directory_organization_nodes_are_provider_scoped_and_atomic(monkeypatch):
    from app.core.system.enterprise_identity import (
        DirectorySynchronizer,
        IdentityPolicyError,
    )
    from app.core.system.identity_repository import MemoryIdentityRepository
    from app.core.system.identity_sessions import SessionManager

    repo = MemoryIdentityRepository()
    sync = DirectorySynchronizer(repo, SessionManager(repo, secret="a-long-test-secret"), clock=lambda: NOW)
    first = sync.apply(provider_uid="idp-a", source="directory", source_event_id="d1", cursor="1",
                       cursor_sequence=1, event_type="DEPARTMENT",
                       attributes={"external_id": "ops", "display_name": "Operations", "action": "UPSERT"})
    sync.apply(provider_uid="idp-a", source="directory", source_event_id="d2", cursor="2",
               cursor_sequence=2, event_type="DEPARTMENT",
               attributes={"external_id": "ops", "action": "DISABLE"})
    restored = sync.apply(provider_uid="idp-a", source="directory", source_event_id="d3", cursor="3",
                          cursor_sequence=3, event_type="DEPARTMENT",
                          attributes={"external_id": "ops", "display_name": "Operations", "action": "RESTORE"})
    assert first["node_uid"] == restored["node_uid"]
    assert repo.get_organization_node("idp-a", "department", "ops")["status"] == "active"
    sync.apply(provider_uid="idp-b", source="directory", source_event_id="d1", cursor="1",
               cursor_sequence=1, event_type="DEPARTMENT",
               attributes={"external_id": "ops", "display_name": "Other Operations", "action": "UPSERT"})
    assert repo.get_organization_node("idp-b", "department", "ops")["display_name"] == "Other Operations"
    with pytest.raises(IdentityPolicyError, match="idempotency"):
        sync.apply(provider_uid="idp-a", source="directory", source_event_id="d3", cursor="3",
                   cursor_sequence=3, event_type="DEPARTMENT",
                   attributes={"external_id": "ops", "display_name": "Changed", "action": "RESTORE"})

    original = repo.put_directory_event
    monkeypatch.setattr(repo, "put_directory_event", lambda event: (_ for _ in ()).throw(RuntimeError("event failure")))
    with pytest.raises(RuntimeError, match="event failure"):
        sync.apply(provider_uid="idp-a", source="rollback", source_event_id="g1", cursor="1",
                   cursor_sequence=1, event_type="GROUP",
                   attributes={"external_id": "stewards", "display_name": "Stewards", "action": "UPSERT"})
    monkeypatch.setattr(repo, "put_directory_event", original)
    assert repo.get_organization_node("idp-a", "group", "stewards") is None
    assert repo.get_directory_checkpoint("idp-a", "rollback") is None


def test_refresh_rotation_reuse_concurrency_risk_and_token_version():
    from app.core.system.enterprise_identity import IdentityPolicyError
    from app.core.system.identity_repository import MemoryIdentityRepository
    from app.core.system.identity_sessions import SessionManager

    repo = MemoryIdentityRepository()
    manager = SessionManager(repo, secret="a-long-test-secret", clock=lambda: NOW, max_sessions=2)
    one = manager.create(provider_uid="idp-1", subject="s1", user_uid="u1", roles=["viewer"], identity_source="oidc")
    rotated = manager.refresh(one.refresh_token)
    with pytest.raises(IdentityPolicyError, match="reuse"):
        manager.refresh(one.refresh_token)
    assert repo.get_session(rotated.session_uid)["status"] == "revoked"
    manager.create(provider_uid="idp-1", subject="s1", user_uid="u1", roles=["viewer"], identity_source="oidc")
    latest = manager.create(provider_uid="idp-1", subject="s1", user_uid="u1", roles=["editor"], identity_source="oidc")
    manager.create(provider_uid="idp-1", subject="s1", user_uid="u1", roles=["viewer"], identity_source="oidc")
    assert len([x for x in repo.list_sessions("idp-1", "s1") if x["status"] == "active"]) <= 2
    manager.risk_revoke(latest.session_uid, reason="impossible_travel")
    assert repo.get_session(latest.session_uid)["risk_reason"] == "impossible_travel"


def test_emergency_access_requires_two_distinct_non_self_approvers_and_window():
    from app.core.system.enterprise_identity import EmergencyAccess, IdentityPolicyError
    from app.core.system.identity_repository import MemoryIdentityRepository
    from app.core.system.identity_sessions import SessionManager

    repo = MemoryIdentityRepository()
    emergency = EmergencyAccess(repo, clock=lambda: NOW)
    with pytest.raises(IdentityPolicyError):
        emergency.request(requester_uid="requester", account_uid="breakglass", reason="IdP outage",
                          expires_at=NOW + timedelta(minutes=30), account_is_local_active_admin=False)
    req = emergency.request(requester_uid="requester", account_uid="breakglass", reason="IdP outage",
                            expires_at=NOW + timedelta(minutes=30), account_is_local_active_admin=True)
    for approver in ("requester", "breakglass"):
        with pytest.raises(IdentityPolicyError):
            emergency.approve(req["uid"], approver_uid=approver)
    emergency.approve(req["uid"], approver_uid="admin-a")
    with pytest.raises(IdentityPolicyError):
        emergency.activate(req["uid"])
    emergency.approve(req["uid"], approver_uid="admin-b")
    active = emergency.activate(req["uid"])
    assert active["status"] == "active"
    manager = SessionManager(repo, secret="a-long-test-secret", clock=lambda: NOW)
    credentials = manager.create(
        subject="breakglass", user_uid="breakglass", roles=["admin"], identity_source="emergency",
        emergency_request_uid=req["uid"], emergency_expires_at=req["expires_at"])
    rotated = manager.refresh(credentials.refresh_token)
    assert repo.get_session(rotated.session_uid)["expires_at"] == req["expires_at"]
    emergency.close(req["uid"], actor_uid="admin-a")
    with pytest.raises(IdentityPolicyError, match="window"):
        manager.refresh(rotated.refresh_token)
    reviewed = emergency.review(req["uid"], reviewer_uid="admin-b", outcome="approved")
    assert reviewed["status"] == "reviewed"


def test_identity_audit_sanitizes_sensitive_fields_recursively():
    from app.core.system.enterprise_identity import sanitize_audit_detail

    safe = sanitize_audit_detail({"code": "secret", "token": "secret", "claims": {"sub": "s1"},
                                  "cookie": "secret", "password": "secret", "assertion": "secret",
                                  "ok": True, "nested": {"verifier": "secret", "event": "login"}})
    serialized = repr(safe)
    assert "secret" not in serialized and "s1" not in serialized
    assert safe["ok"] is True and safe["nested"]["event"] == "login"


def test_public_identity_policy_is_method_exact_and_logout_is_viewer_safe():
    from app.core.system.permissions import (
        IDENTITY_READ,
        MANAGE_USERS,
        PUBLIC,
        permission_for_request,
    )

    assert permission_for_request("/api/system/identity/providers", "GET") == (PUBLIC,)
    assert permission_for_request("/api/system/identity/providers", "POST") != (PUBLIC,)
    assert permission_for_request("/api/system/identity/callback", "POST") != (PUBLIC,)
    assert permission_for_request("/api/system/auth/login", "GET") == (MANAGE_USERS,)
    assert permission_for_request("/api/system/identity/logout", "POST") == (IDENTITY_READ,)
    assert permission_for_request("/api/system/identity/idp-versions", "GET") == ("identity:manage",)
    assert permission_for_request("/api/system/identity/sessions", "GET") == ("identity:operate",)
    assert permission_for_request("/api/system/identity/audit", "GET") == ("identity:manage",)


def test_identity_management_input_helpers_reject_malformed_values():
    from app.api.system.enterprise_identity import _positive_int, _timestamp, _uuid
    from app.core.system.enterprise_identity import IdentityPolicyError

    for value in (None, "not-a-uuid", 123):
        with pytest.raises(IdentityPolicyError):
            _uuid(value, "provider_uid")
    for value in (None, "bad", 0, -1, True, 501):
        with pytest.raises(IdentityPolicyError):
            _positive_int(value, "limit", maximum=500)
    for value in (None, "2026-08-02", "not-a-date"):
        with pytest.raises(IdentityPolicyError):
            _timestamp(value, "expires_at")


def test_identity_management_api_returns_safe_4xx_and_upstream_502(monkeypatch):
    from app import create_app
    from app.api.system import enterprise_identity as identity_api
    from app.core.system.enterprise_identity import IdentityUpstreamError
    from app.core.system.oidc import OidcTransport

    monkeypatch.setattr(
        "app.core.system.auth.load_identity_from_token",
        lambda token, secret: {"id": "01900000-0000-7000-8000-000000000001", "username": "admin",
                               "roles": ["admin"]} if token == "admin" else None,
    )
    monkeypatch.setattr(identity_api, "_audit", lambda *args, **kwargs: None)
    app = create_app()
    app.config.update(TESTING=True)
    client = app.test_client()
    headers = {"Authorization": "Bearer admin"}
    malformed = (
        client.post("/api/system/identity/idp-versions", json={"provider_uid": "bad"}, headers=headers),
        client.get("/api/system/identity/sessions?provider_uid=bad", headers=headers),
        client.get("/api/system/identity/audit?limit=not-an-int", headers=headers),
        client.post("/api/system/identity/directory/delta", json={"provider_uid": "bad"}, headers=headers),
        client.post("/api/system/identity/emergency/requests", json={"account_uid": "bad"}, headers=headers),
    )
    assert all(response.status_code == 400 for response in malformed)
    assert all("Traceback" not in response.get_data(as_text=True) for response in malformed)

    provider_uid = "01900000-0000-7000-8000-000000000099"
    monkeypatch.setattr(identity_api, "_load_config", lambda *args, **kwargs: {
        "provider_uid": provider_uid, "version": 1, "status": "draft",
        "issuer": "https://login.example.com/tenant", "client_id": "dataops",
        "secret_ref": "env:DATAOPS_OIDC_TEST_SECRET",
        "authorization_endpoint": "https://login.example.com/authorize",
        "token_endpoint": "https://login.example.com/token", "jwks_uri": "https://login.example.com/jwks",
        "redirect_uris": ["https://dataops.example.com/callback"], "algorithms": ["RS256"],
        "mapping_version": "v1", "config": {
            "post_login_redirect_uris": ["https://dataops.example.com/login/callback"],
            "group_rules": {"stewards": {"roles": ["viewer"]}},
        },
    })
    monkeypatch.setattr(OidcTransport, "fetch_jwks", lambda self, config: (_ for _ in ()).throw(
        IdentityUpstreamError("internal network detail must not leak")))
    response = client.post(f"/api/system/identity/idp-versions/{provider_uid}/1/connectivity",
                           json={}, headers=headers)
    assert response.status_code == 502
    assert "internal network detail" not in response.get_data(as_text=True)