import json
import typing
from typing import Any

import httpx


class KeycloakAPIClient:  # noqa: PLR0904
    KEYCLOAK_MAIN_REALM = "master"

    def __init__(
        self,
        base_url: str,
        master_user: str,
        master_pass: str,
        public_url: str | None = None,
        legacy: bool = False,
    ) -> None:
        self._client = httpx.Client()
        self.base_url = base_url
        self._master_user = master_user
        self._master_pass = master_pass
        self._public_url = public_url

        if self._public_url is None:
            self._public_url = self.base_url

        self.legacy = legacy

        main_realm_auth_endpoint = self.get_auth_endpoint_for_realm(
            self.endpoint, self.KEYCLOAK_MAIN_REALM, legacy=self.legacy
        )
        self.master_token, self._refresh_token = self.get_tokens_for_user(
            realm_auth_endpoint=main_realm_auth_endpoint,
            username=self._master_user,
            password=self._master_pass,
            client_id="admin-cli",
        )

        # Caching
        self._realm_client_cache: dict[str, dict[str, str]] = {}

    @property
    def endpoint(self) -> str:
        legacy_auth_suffix = "/auth" if self.legacy else ""
        return f"{self.base_url}{legacy_auth_suffix}"

    @property
    def auth_header(self) -> dict[str, str]:
        endpoint = self.get_auth_endpoint_for_realm(
            endpoint=self.endpoint, realm=self.KEYCLOAK_MAIN_REALM, legacy=self.legacy
        )
        try:
            self.master_token = self.refresh_access_token(
                realm_auth_endpoint=endpoint, refresh_token=self._refresh_token, client_id="admin-cli"
            )
        except httpx.RequestError:  # If refresh token is expired
            self.master_token, self._refresh_token = self.get_tokens_for_user(
                realm_auth_endpoint=endpoint,
                username=self._master_user,
                password=self._master_pass,
                client_id="admin-cli",
            )
        return {"Authorization": f"Bearer {self.master_token}"}

    def get_tokens_for_user(
        self, realm_auth_endpoint: str, username: str, password: str, client_id: str = "admin-cli"
    ) -> tuple[str, str]:
        response = self._client.post(
            realm_auth_endpoint,
            data={"client_id": client_id, "username": username, "password": password, "grant_type": "password"},
        )
        response.raise_for_status()
        master_tokens = response.json()
        return master_tokens.get("access_token"), master_tokens.get("refresh_token")

    def refresh_access_token(self, realm_auth_endpoint: str, refresh_token: str, client_id: str) -> str:
        response = self._client.post(
            realm_auth_endpoint,
            data={"client_id": client_id, "grant_type": "refresh_token", "refresh_token": refresh_token},
        )

        if response.status_code == 200:
            response = response.json()
            return response.get("access_token")
        msg = "Unauthorized, refresh token is invalid or expired"
        raise httpx.RequestError(msg)

    def get_auth_endpoint_for_realm(self, endpoint: str, realm: str, legacy: bool = False):
        endpoint = endpoint + "/auth" if legacy else endpoint
        return f"{endpoint}/realms/{realm}/protocol/openid-connect/token"

    def get_current_user_info(self, realm: str) -> dict[str, Any]:
        endpoint = f"{self.endpoint}/realms/{realm}/protocol/openid-connect/userinfo"
        response = self._client.get(endpoint, headers=self.auth_header)
        response.raise_for_status()
        return response.json()

    def create_realm(self, realm: str) -> bool:
        endpoint = f"{self.endpoint}/admin/realms"
        response = self._client.post(
            endpoint,
            headers=self.auth_header,
            json={
                "realm": realm,
                "displayName": "CSIRT MU Test realm",
                "enabled": True,
                "sslRequired": "external",
                "registrationAllowed": True,
                "loginWithEmailAllowed": True,
                "resetPasswordAllowed": True,
                "editUsernameAllowed": False,
                "bruteForceProtected": True,
                "attributes": {
                    "cibaBackchannelTokenDeliveryMode": "poll",
                    "cibaAuthRequestedUserHint": "login_hint",
                    "oauth2DevicePollingInterval": "5",
                    "clientOfflineSessionMaxLifespan": "0",
                    "clientSessionIdleTimeout": "0",
                    "userProfileEnabled": "false",
                    "clientOfflineSessionIdleTimeout": "0",
                    "cibaInterval": "5",
                    "cibaExpiresIn": "120",
                    "oauth2DeviceCodeLifespan": "600",
                    "parRequestUriLifespan": "60",
                    "clientSessionMaxLifespan": "0",
                    "frontendUrl": self._public_url,
                },
            },
        )
        response.raise_for_status()
        return True

    def create_client(self, realm: str, client_name: str, client_secret: str) -> bool:
        endpoint = f"{self.endpoint}/admin/realms/{realm}/clients"
        response = self._client.post(
            endpoint,
            headers=self.auth_header,
            json={
                "clientId": client_name,
                "name": client_name,
                "enabled": True,
                "protocol": "openid-connect",
                "secret": client_secret,
                "publicClient": False,
                "clientAuthenticatorType": "client-secret",
                "authorizationServicesEnabled": True,
                "serviceAccountsEnabled": True,
            },
        )
        response.raise_for_status()
        return True

    def set_redirect_uris_for_client(self, realm: str, client_name: str, redirect_uris: list[str]) -> bool:
        client_id = self.find_client_id_in_realm(realm=realm, client_name=client_name)
        endpoint = f"{self.endpoint}/admin/realms/{realm}/clients/{client_id}"
        response = self._client.put(endpoint, headers=self.auth_header, json={"redirectUris": redirect_uris})
        return response.status_code == 204

    def set_post_logout_redirect_uris_for_client(self, realm: str, client_name: str, redirect_uris: list[str]) -> bool:
        client_id = self.find_client_id_in_realm(realm=realm, client_name=client_name)
        endpoint = f"{self.endpoint}/admin/realms/{realm}/clients/{client_id}"
        response = self._client.put(
            endpoint,
            headers=self.auth_header,
            json={"attributes": {"post.logout.redirect.uris": "##".join(redirect_uris)}},
        )
        return response.status_code == 204

    def set_web_origins_for_client(self, realm: str, client_name: str, web_origins: list[str]) -> bool:
        client_id = self.find_client_id_in_realm(realm=realm, client_name=client_name)
        endpoint = f"{self.endpoint}/admin/realms/{realm}/clients/{client_id}"
        response = self._client.put(endpoint, headers=self.auth_header, json={"webOrigins": web_origins})
        return response.status_code == 204

    def get_clients(self, realm: str) -> list[dict[str, Any]]:
        endpoint = f"{self.endpoint}/admin/realms/{realm}/clients"
        response = self._client.get(endpoint, headers=self.auth_header)
        response.raise_for_status()
        return response.json()

    def create_client_scope(self, realm: str, name: str) -> bool:
        endpoint = f"{self.endpoint}/admin/realms/{realm}/client-scopes"
        response = self._client.post(
            endpoint,
            headers=self.auth_header,
            json={
                "name": name,
                "protocol": "openid-connect",
            },
        )
        response.raise_for_status()
        return True

    def make_client_scope_default(self, realm: str, client_scope_id: str) -> bool:
        # This is how keycloak does it man
        scope_endpoint = f"{self.endpoint}/admin/realms/{realm}/client-scopes/{client_scope_id}"
        scope = self._client.get(url=scope_endpoint, headers=self.auth_header).json()
        scope["type"] = "default"
        put = self._client.put(
            url=f"{self.endpoint}/admin/realms/{realm}/client-scopes/{client_scope_id}",
            json=scope,
            headers=self.auth_header,
        )
        put.raise_for_status()

        # My brother in christ why default-default
        endpoint = f"{self.endpoint}/admin/realms/{realm}/default-default-client-scopes/{client_scope_id}"
        delete = self._client.delete(url=endpoint, headers=self.auth_header)
        delete.raise_for_status()
        put = self._client.put(endpoint, headers=self.auth_header)
        put.raise_for_status()
        return True

    def get_client_scopes(self, realm: str) -> list[dict[str, Any]]:
        endpoint = f"{self.endpoint}/admin/realms/{realm}/client-scopes"
        response = self._client.get(endpoint, headers=self.auth_header)
        response.raise_for_status()
        return response.json()

    def add_default_client_scope_to_client(self, realm: str, client_name: str, client_scope_id: str):
        client_id = self.find_client_id_in_realm(realm=realm, client_name=client_name)
        endpoint = f"{self.endpoint}/admin/realms/{realm}/clients/{client_id}/default-client-scopes/{client_scope_id}"
        response = self._client.put(
            endpoint,
            headers=self.auth_header,
            data={"client": client_id, "clientScopeId": client_scope_id, "realm": realm},
        )
        response.raise_for_status()
        return True

    def remove_default_client_scope_to_client(self, realm: str, client_name: str, client_scope_id: str):
        client_id = self.find_client_id_in_realm(realm=realm, client_name=client_name)
        endpoint = f"{self.endpoint}/admin/realms/{realm}/clients/{client_id}/default-client-scopes/{client_scope_id}"
        response = self._client.request(
            method="DELETE",
            url=endpoint,
            headers=self.auth_header,
            data={"client": client_id, "clientScopeId": client_scope_id, "realm": realm},
        )
        response.raise_for_status()
        return True

    def add_optional_client_scope_to_client(self, realm: str, client_name: str, client_scope_id: str):
        client_id = self.find_client_id_in_realm(realm=realm, client_name=client_name)
        endpoint = f"{self.endpoint}/admin/realms/{realm}/clients/{client_id}/optional-client-scopes/{client_scope_id}"
        response = self._client.put(
            endpoint,
            headers=self.auth_header,
            data={"client": client_id, "clientScopeId": client_scope_id, "realm": realm},
        )
        response.raise_for_status()
        return True

    def remove_optional_client_scope_to_client(self, realm: str, client_name: str, client_scope_id: str):
        client_id = self.find_client_id_in_realm(realm=realm, client_name=client_name)
        endpoint = f"{self.endpoint}/admin/realms/{realm}/clients/{client_id}/optional-client-scopes/{client_scope_id}"
        response = self._client.request(
            method="DELETE",
            url=endpoint,
            headers=self.auth_header,
            data={"client": client_id, "clientScopeId": client_scope_id, "realm": realm},
        )
        response.raise_for_status()
        return True

    def add_oidc_claim_to_client(
        self,
        realm: str,
        client_name: str,
        mapper_name: str,
        claim_name: str,
        claim_value: str | dict[str, Any] | list[Any],
    ):
        client_id = self.find_client_id_in_realm(realm=realm, client_name=client_name)
        endpoint = f"{self.endpoint}/admin/realms/{realm}/clients/{client_id}/protocol-mappers/models"
        if not isinstance(claim_value, str):
            claim_value = json.dumps(claim_value)
        response = self._client.post(
            endpoint,
            headers=self.auth_header,
            json={
                "protocol": "openid-connect",
                "config": {
                    "id.token.claim": True,
                    "access.token.claim": True,
                    "userinfo.token.claim": True,
                    "access.tokenResponse.claim": False,
                    "jsonType.label": "JSON",
                    "claim.name": claim_name,
                    "claim.value": claim_value,
                },
                "name": mapper_name,
                "protocolMapper": "oidc-hardcoded-claim-mapper",
            },
        )
        response.raise_for_status()
        return True

    def add_oidc_claim_to_client_scope(
        self,
        realm: str,
        client_scope_id: str,
        mapper_name: str,
        claim_name: str,
        claim_value: str | list[str] | dict[str, Any],
    ):
        endpoint = f"{self.endpoint}/admin/realms/{realm}/client-scopes/{client_scope_id}/protocol-mappers/models"
        if not isinstance(claim_value, str):
            claim_value = json.dumps(claim_value)
        response = self._client.post(
            endpoint,
            headers=self.auth_header,
            json={
                "protocol": "openid-connect",
                "config": {
                    "id.token.claim": True,
                    "access.token.claim": True,
                    "userinfo.token.claim": True,
                    "access.tokenResponse.claim": False,
                    "jsonType.label": "JSON",
                    "claim.name": claim_name,
                    "claim.value": claim_value,
                },
                "name": mapper_name,
                "protocolMapper": "oidc-hardcoded-claim-mapper",
            },
        )
        response.raise_for_status()
        return True

    def create_user(
        self, username: str, password: str, realm: str, first_name: str, last_name: str, email: str
    ) -> bool:
        endpoint = f"{self.endpoint}/admin/realms/{realm}/users"
        response = self._client.post(
            endpoint,
            headers=self.auth_header,
            json={
                "username": username,
                "email": email,
                "firstName": first_name,
                "lastName": last_name,
                "credentials": [{"value": password, "type": "password", "temporary": False}],
                "emailVerified": True,
                "enabled": True,
            },
        )
        response.raise_for_status()
        return True

    def get_roles_for_realm(self, realm: str):
        endpoint = f"{self.endpoint}/admin/realms/{realm}/roles"
        response = self._client.get(endpoint, headers=self.auth_header)
        response.raise_for_status()
        return response.json()

    def get_users_in_realm(self, realm: str):
        endpoint = f"{self.endpoint}/admin/realms/{realm}/users"
        response = self._client.get(endpoint, headers=self.auth_header)
        response.raise_for_status()
        return response.json()

    def get_user_id_in_realm_by_username(self, realm: str, username: str) -> str | None:
        users = self.get_users_in_realm(realm)
        for user in users:
            if user["username"] == username:
                return user["id"]
        return None

    def remove_role_mapping_from_user(self, realm: str, username: str, role_name: str):
        user_id = self.get_user_id_in_realm_by_username(realm=realm, username=username)
        if not user_id:
            msg = f"User {username} not found in realm {realm}!"
            raise ValueError(msg)

        roles = self.get_roles_for_realm(realm)
        role_id = None
        for role in roles:
            if role["name"] == role_name:
                role_id = role["id"]
        if not role_id:
            msg = f"Role {role_name} not found in realm {realm}!"
            raise ValueError(msg)

        endpoint = f"{self.endpoint}/admin/realms/{realm}/users/{user_id}/role-mappings/realm"
        response = self._client.request(
            method="DELETE", url=endpoint, headers=self.auth_header, json=[{"id": role_id, "name": role_name}]
        )

        response.raise_for_status()
        return True

    def add_role_mapping_to_user(self, realm: str, username: str, role_name: str):
        user_id = self.get_user_id_in_realm_by_username(realm=realm, username=username)
        if not user_id:
            msg = f"User {username} not found in realm {realm}!"
            raise ValueError(msg)

        roles = self.get_roles_for_realm(realm)
        role_id = None
        for role in roles:
            if role["name"] == role_name:
                role_id = role["id"]
        if not role_id:
            msg = f"Role {role_name} not found in realm {realm}!"
            raise ValueError(msg)

        endpoint = f"{self.endpoint}/admin/realms/{realm}/users/{user_id}/role-mappings/realm"
        response = self._client.post(endpoint, headers=self.auth_header, json=[{"id": role_id, "name": role_name}])

        response.raise_for_status()
        return True

    def find_client_id_in_realm(self, realm: str, client_name: str) -> str:
        if realm in self._realm_client_cache and (client_id := self._realm_client_cache[realm].get(client_name)):
            return client_id
        if realm not in self._realm_client_cache:
            self._realm_client_cache[realm] = {}

        endpoint = f"{self.endpoint}/admin/realms/{realm}/clients"
        response = self._client.get(url=endpoint, headers=self.auth_header, params={"clientId": client_name})
        response.raise_for_status()
        response = response.json()

        if response:
            client_id = response[0].get("id")
            self._realm_client_cache[realm][client_name] = client_id
            return client_id

        msg = f"Client: {client_name} not found in realm {realm}"
        raise ValueError(msg)

    def _enable_experimental_permissions_on_client(self, realm: str, client_name: str) -> bool:
        client_id = self.find_client_id_in_realm(realm=realm, client_name=client_name)
        endpoint = f"{self.endpoint}/admin/realms/{realm}/clients/{client_id}/management/permissions"
        response = self._client.put(endpoint, headers=self.auth_header, json={"enabled": True})
        response.raise_for_status()
        return True

    def create_client_policy(
        self, realm: str, name: str, description: str, client_names: list[str], logic: str = "positive"
    ):
        # Notes
        # 1. Decision strategy for the Client Policies seems to always be UNANIMOUS
        # 2. Client in URL is the "realm_management" client

        client_ids = [
            self.find_client_id_in_realm(realm=realm, client_name=client_name) for client_name in client_names
        ]
        realm_management_client_id = self.find_client_id_in_realm(realm=realm, client_name="realm-management")

        endpoint = (
            f"{self.endpoint}/admin/realms/{realm}/clients/{realm_management_client_id}"
            f"/authz/resource-server/policy/client"
        )

        response = self._client.post(
            endpoint,
            headers=self.auth_header,
            json={
                "clients": client_ids,
                "decisionStrategy": "UNANIMOUS",
                "name": name,
                "description": description,
                "logic": logic,
            },
        )
        response.raise_for_status()
        return True

    def _find_client_token_exchange_permission_in_realm(self, realm: str, client_name: str) -> dict[str, Any]:
        client_id = self.find_client_id_in_realm(realm=realm, client_name=client_name)
        management_client_id = self.find_client_id_in_realm(realm=realm, client_name="realm-management")
        endpoint = (
            f"{self.endpoint}/admin/realms/{realm}/clients/{management_client_id}"
            f"/authz/resource-server/permission/scope/"
        )

        response = self._client.get(
            endpoint,
            headers=self.auth_header,
        )
        response.raise_for_status()

        token_exchange_policy_name = f"token-exchange.permission.client.{client_id}"
        permission = next((item for item in response.json() if item["name"] == token_exchange_policy_name), None)

        if not permission:
            msg = f"Token-exchange permission not found in realm: {realm} management."
            raise ValueError(msg)
        return permission

    def _find_policy_by_name(self, realm: str, policy_name: str) -> typing.Any:
        management_client_id = self.find_client_id_in_realm(realm=realm, client_name="realm-management")
        endpoint = (
            f"{self.endpoint}/admin/realms/{realm}/clients/{management_client_id}/authz/resource-server/policy/client"
        )
        response = self._client.get(
            endpoint,
            headers=self.auth_header,
        )
        policy = next((item for item in response.json() if item["name"] == policy_name), None)

        if not policy:
            msg = f"Token-exchange policy not found in realm: {realm} management."
            raise ValueError(msg)
        return policy

    def _find_client_resource_id(self, realm: str, target_client_name: str) -> str:
        management_client_id = self.find_client_id_in_realm(realm=realm, client_name="realm-management")
        target_client_id = self.find_client_id_in_realm(realm=realm, client_name=target_client_name)
        endpoint = f"{self.endpoint}/admin/realms/{realm}/clients/{management_client_id}/authz/resource-server/resource"
        response = self._client.get(
            endpoint,
            headers=self.auth_header,
        )
        resource = next(
            (item for item in response.json() if item["name"] == f"client.resource.{target_client_id}"), None
        )
        if not resource:
            msg = f"Client {target_client_name} resource not found in realm {realm} management."
            raise ValueError(msg)
        return resource["_id"]

    def _find_scope_in_realm(self, realm: str, scope_name: str) -> str | None:
        management_client_id = self.find_client_id_in_realm(realm=realm, client_name="realm-management")
        endpoint = f"{self.endpoint}/admin/realms/{realm}/clients/{management_client_id}/authz/resource-server/scope"
        response = self._client.get(
            endpoint,
            headers=self.auth_header,
        )
        scope = next((item for item in response.json() if item["name"] == scope_name), None)
        if not scope:
            msg = f"Scope {scope_name} not found in realm {realm} management."
            raise ValueError(msg)
        return scope["id"]

    def enable_token_exchange_from_client_to_client(self, realm: str, source_client_name: str, target_client_name: str):
        # Enable permissions on clients
        for client_name in [source_client_name, target_client_name]:
            self._enable_experimental_permissions_on_client(realm=realm, client_name=client_name)

        # Create token exchange policy
        exchange_policy_name = f"token-exchange-from-{source_client_name}-client"
        self.create_client_policy(
            realm=realm,
            name=exchange_policy_name,
            description=f"Allow token-exchange from client {source_client_name}",
            client_names=[source_client_name],
            logic="POSITIVE",
        )

        # Find management client
        management_client_id = self.find_client_id_in_realm(realm=realm, client_name="realm-management")

        # Find auto created token exchange permission
        exchange_permission = self._find_client_token_exchange_permission_in_realm(
            realm=realm, client_name=target_client_name
        )

        # Update auto generated permission scope with configured token exchange policy
        endpoint = (
            f"{self.endpoint}/admin/realms/{realm}/clients/{management_client_id}"
            f"/authz/resource-server/permission/scope/{exchange_permission['id']}"
        )
        response = self._client.put(
            endpoint,
            headers=self.auth_header,
            json={
                "decisionStrategy": "UNANIMOUS",
                "id": exchange_permission["id"],
                "logic": "POSITIVE",
                "name": exchange_permission["name"],
                "policies": [self._find_policy_by_name(realm, exchange_policy_name)["id"]],
                "resources": [self._find_client_resource_id(realm=realm, target_client_name=target_client_name)],
                "scopes": [self._find_scope_in_realm(realm=realm, scope_name="token-exchange")],
                "type": "scope",
            },
        )
        response.raise_for_status()
        return True

    def create_group_in_realm(self, realm: str, group_name: str) -> bool:
        response = self._client.post(
            url=f"{self.endpoint}/admin/realms/{realm}/groups", headers=self.auth_header, json={"name": group_name}
        )
        response.raise_for_status()
        return httpx.codes.is_success(response.status_code)

    def get_realm_group_by_name(self, realm: str, group_name: str) -> str:
        response = self._client.get(
            url=f"{self.endpoint}/admin/realms/{realm}/groups",
            headers=self.auth_header,
            params={"search": group_name, "exact": True},
        )
        response.raise_for_status()
        return response.json()[0]["id"]

    def create_role_in_realm(self, realm: str, role_name: str, role_description: str | None = None) -> bool:
        payload = {"name": role_name}
        if role_description is not None:
            payload["description"] = role_description

        response = self._client.post(
            url=f"{self.endpoint}/admin/realms/{realm}/roles", headers=self.auth_header, json=payload
        )
        response.raise_for_status()
        return httpx.codes.is_success(response.status_code)

    def get_realm_role_by_name(self, realm: str, role_name: str) -> str:
        response = self._client.get(
            url=f"{self.endpoint}/admin/realms/{realm}/roles",
            headers=self.auth_header,
            params={"search": role_name, "exact": True},
        )
        response.raise_for_status()
        return response.json()[0]["id"]

    def add_role_to_group(self, realm: str, group_id: str, role_id: str, role_name: str) -> bool:
        response = self._client.post(
            url=f"{self.endpoint}/admin/realms/{realm}/groups/{group_id}/role-mappings/realm",
            headers=self.auth_header,
            json=[{"id": role_id, "name": role_name}],
        )
        response.raise_for_status()
        return httpx.codes.is_success(response.status_code)

    def add_user_to_group(self, realm: str, user_id: str, group_id: str) -> bool:
        response = self._client.put(
            url=f"{self.endpoint}/admin/realms/{realm}/users/{user_id}/groups/{group_id}",
            headers=self.auth_header,
        )
        response.raise_for_status()
        return httpx.codes.is_success(response.status_code)

    def add_realm_roles_mapping_to_client_scope(self, realm: str, client_scope_id: str, client_scope_name: str):
        response = self._client.post(
            url=f"{self.endpoint}/admin/realms/{realm}/client-scopes/{client_scope_id}/protocol-mappers/add-models",
            headers=self.auth_header,
            json=[
                {
                    "consentRequired": False,
                    "name": "realm roles",
                    "protocol": "openid-connect",
                    "protocolMapper": "oidc-usermodel-realm-role-mapper",
                    "config": {
                        "access.token.claim": True,
                        "userinfo.token.claim": True,
                        "claim.name": client_scope_name,
                        "jsonType.label": "String",
                        "multivalued": True,
                        "user.attribute": "foo",  # Ok keycloak
                    },
                }
            ],
        )
        response.raise_for_status()
        return httpx.codes.is_success(response.status_code)

    def create_audience_mapper(self, client_id: str, client_name: str, realm_name: str) -> None:
        response = self._client.post(
            url=f"{self.endpoint}/admin/realms/{realm_name}/clients/{client_id}/protocol-mappers/models",
            headers=self.auth_header,
            json={
                "name": "Audience Mapper",
                "protocol": "openid-connect",
                "protocolMapper": "oidc-audience-mapper",
                "config": {
                    "included.client.audience": client_name,
                    "id.token.claim": False,
                    "access.token.claim": True,
                },
            },
        )
        response.raise_for_status()
