"""Test token view"""

from base64 import b64encode
from json import loads
from urllib.parse import quote

from django.test import RequestFactory
from django.urls import reverse
from jwt import decode

from authentik.blueprints.tests import apply_blueprint
from authentik.common.oauth.constants import (
    GRANT_TYPE_CLIENT_CREDENTIALS,
    GRANT_TYPE_PASSWORD,
    SCOPE_OPENID,
    SCOPE_OPENID_EMAIL,
    SCOPE_OPENID_PROFILE,
    TOKEN_TYPE,
)
from authentik.core.models import Application, Group, Token, TokenIntents, UserTypes
from authentik.core.tests.utils import create_test_admin_user, create_test_cert, create_test_flow
from authentik.policies.models import PolicyBinding
from authentik.providers.oauth2.errors import TokenError
from authentik.providers.oauth2.models import (
    OAuth2Provider,
    RedirectURI,
    RedirectURIMatchingMode,
    ScopeMapping,
)
from authentik.providers.oauth2.tests.utils import OAuthTestCase


class TestTokenClientCredentialsStandardCompat(OAuthTestCase):
    """Test token (client_credentials) view"""

    @apply_blueprint("system/providers-oauth2.yaml")
    def setUp(self) -> None:
        super().setUp()
        self.factory = RequestFactory()
        self.provider = OAuth2Provider.objects.create(
            name="test",
            authorization_flow=create_test_flow(),
            redirect_uris=[RedirectURI(RedirectURIMatchingMode.STRICT, "http://testserver")],
            signing_key=create_test_cert(),
        )
        self.provider.property_mappings.set(ScopeMapping.objects.all())
        self.app = Application.objects.create(name="test", slug="test", provider=self.provider)
        self.user = create_test_admin_user("sa")
        self.user.type = UserTypes.SERVICE_ACCOUNT
        self.user.save()
        self.token = Token.objects.create(
            identifier="sa-token",
            user=self.user,
            intent=TokenIntents.INTENT_APP_PASSWORD,
            expiring=False,
        )

    def test_wrong_user(self):
        """test invalid username"""
        response = self.client.post(
            reverse("authentik_providers_oauth2:token"),
            {
                "grant_type": GRANT_TYPE_CLIENT_CREDENTIALS,
                "scope": SCOPE_OPENID,
                "client_id": self.provider.client_id,
                "client_secret": b64encode(f"saa:{self.token.key}".encode()).decode(),
            },
        )
        self.assertEqual(response.status_code, 400)
        self.assertJSONEqual(
            response.content.decode(),
            {
                "error": "invalid_grant",
                "error_description": TokenError.errors["invalid_grant"],
                "request_id": response.headers["X-authentik-id"],
            },
        )

    def test_wrong_token(self):
        """test invalid token"""
        response = self.client.post(
            reverse("authentik_providers_oauth2:token"),
            {
                "grant_type": GRANT_TYPE_CLIENT_CREDENTIALS,
                "scope": SCOPE_OPENID,
                "client_id": self.provider.client_id,
                "client_secret": b64encode(f"sa:{self.token.key}foo".encode()).decode(),
            },
        )
        self.assertEqual(response.status_code, 400)
        self.assertJSONEqual(
            response.content.decode(),
            {
                "error": "invalid_grant",
                "error_description": TokenError.errors["invalid_grant"],
                "request_id": response.headers["X-authentik-id"],
            },
        )

    def test_no_provider(self):
        """test no provider"""
        self.app.provider = None
        self.app.save()
        response = self.client.post(
            reverse("authentik_providers_oauth2:token"),
            {
                "grant_type": GRANT_TYPE_CLIENT_CREDENTIALS,
                "scope": SCOPE_OPENID,
                "client_id": self.provider.client_id,
                "client_secret": b64encode(f"sa:{self.token.key}".encode()).decode(),
            },
        )
        self.assertEqual(response.status_code, 400)
        self.assertJSONEqual(
            response.content.decode(),
            {
                "error": "invalid_grant",
                "error_description": TokenError.errors["invalid_grant"],
                "request_id": response.headers["X-authentik-id"],
            },
        )

    def test_permission_denied(self):
        """test permission denied"""
        group = Group.objects.create(name="foo")
        PolicyBinding.objects.create(
            group=group,
            target=self.app,
            order=0,
        )
        response = self.client.post(
            reverse("authentik_providers_oauth2:token"),
            {
                "grant_type": GRANT_TYPE_CLIENT_CREDENTIALS,
                "scope": SCOPE_OPENID,
                "client_id": self.provider.client_id,
                "client_secret": b64encode(f"sa:{self.token.key}".encode()).decode(),
            },
        )
        self.assertEqual(response.status_code, 400)
        self.assertJSONEqual(
            response.content.decode(),
            {
                "error": "invalid_grant",
                "error_description": TokenError.errors["invalid_grant"],
                "request_id": response.headers["X-authentik-id"],
            },
        )

    def test_successful(self):
        """test successful"""
        response = self.client.post(
            reverse("authentik_providers_oauth2:token"),
            {
                "grant_type": GRANT_TYPE_CLIENT_CREDENTIALS,
                "scope": f"{SCOPE_OPENID} {SCOPE_OPENID_EMAIL} {SCOPE_OPENID_PROFILE}",
                "client_id": self.provider.client_id,
                "client_secret": b64encode(f"sa:{self.token.key}".encode()).decode(),
            },
        )
        self.assertEqual(response.status_code, 200)
        body = loads(response.content.decode())
        self.assertEqual(body["token_type"], TOKEN_TYPE)
        _, alg = self.provider.jwt_key
        jwt = decode(
            body["access_token"],
            key=self.provider.signing_key.public_key,
            algorithms=[alg],
            audience=self.provider.client_id,
        )
        self.assertEqual(jwt["given_name"], self.user.name)
        self.assertEqual(jwt["preferred_username"], self.user.username)
        jwt = decode(
            body["id_token"],
            key=self.provider.signing_key.public_key,
            algorithms=[alg],
            audience=self.provider.client_id,
        )
        self.assertEqual(jwt["given_name"], self.user.name)
        self.assertEqual(jwt["preferred_username"], self.user.username)

    def test_successful_basic_auth_urlencoded_client_secret(self):
        """test successful with URL-encoded Basic auth credentials"""
        client_secret = b64encode(f"sa:{self.token.key}".encode()).decode()
        header = b64encode(
            f"{quote(self.provider.client_id, safe='')}:{quote(client_secret, safe='')}".encode()
        ).decode()
        response = self.client.post(
            reverse("authentik_providers_oauth2:token"),
            {
                "grant_type": GRANT_TYPE_CLIENT_CREDENTIALS,
                "scope": f"{SCOPE_OPENID} {SCOPE_OPENID_EMAIL} {SCOPE_OPENID_PROFILE}",
            },
            HTTP_AUTHORIZATION=f"Basic {header}",
        )
        self.assertEqual(response.status_code, 200)
        body = loads(response.content.decode())
        self.assertEqual(body["token_type"], TOKEN_TYPE)
        _, alg = self.provider.jwt_key
        jwt = decode(
            body["access_token"],
            key=self.provider.signing_key.public_key,
            algorithms=[alg],
            audience=self.provider.client_id,
        )
        self.assertEqual(jwt["given_name"], self.user.name)
        self.assertEqual(jwt["preferred_username"], self.user.username)
        jwt = decode(
            body["id_token"],
            key=self.provider.signing_key.public_key,
            algorithms=[alg],
            audience=self.provider.client_id,
        )
        self.assertEqual(jwt["given_name"], self.user.name)
        self.assertEqual(jwt["preferred_username"], self.user.username)

    def test_successful_password(self):
        """test successful (password grant)"""
        response = self.client.post(
            reverse("authentik_providers_oauth2:token"),
            {
                "grant_type": GRANT_TYPE_PASSWORD,
                "scope": f"{SCOPE_OPENID} {SCOPE_OPENID_EMAIL} {SCOPE_OPENID_PROFILE}",
                "client_id": self.provider.client_id,
                "client_secret": b64encode(f"sa:{self.token.key}".encode()).decode(),
            },
        )
        self.assertEqual(response.status_code, 200)
        body = loads(response.content.decode())
        self.assertEqual(body["token_type"], TOKEN_TYPE)
        _, alg = self.provider.jwt_key
        jwt = decode(
            body["access_token"],
            key=self.provider.signing_key.public_key,
            algorithms=[alg],
            audience=self.provider.client_id,
        )
        self.assertEqual(jwt["given_name"], self.user.name)
        self.assertEqual(jwt["preferred_username"], self.user.username)
