"""Crypto tests"""

from json import loads
from os import makedirs
from pathlib import Path
from tempfile import TemporaryDirectory

from cryptography.x509.extensions import SubjectAlternativeName
from cryptography.x509.general_name import DNSName
from django.urls import reverse
from django.utils.timezone import now
from rest_framework.test import APITestCase

from authentik.core.api.used_by import DeleteAction
from authentik.core.tests.utils import (
    create_test_admin_user,
    create_test_cert,
    create_test_flow,
    create_test_user,
)
from authentik.crypto.api import CertificateKeyPairSerializer
from authentik.crypto.builder import CertificateBuilder
from authentik.crypto.models import CertificateKeyPair, generate_key_id, generate_key_id_legacy
from authentik.crypto.tasks import MANAGED_DISCOVERED, certificate_discovery
from authentik.lib.config import CONFIG
from authentik.lib.generators import generate_id, generate_key
from authentik.providers.oauth2.models import OAuth2Provider, RedirectURI, RedirectURIMatchingMode


class TestCrypto(APITestCase):
    """Test Crypto validation"""

    def test_model_private(self):
        """Test model private key"""
        cert = CertificateKeyPair.objects.create(
            name=generate_id(),
            certificate_data="foo",
            key_data="foo",
        )
        self.assertIsNone(cert.private_key)

    def test_serializer(self):
        """Test API Validation"""
        keypair = create_test_cert()
        self.assertTrue(
            CertificateKeyPairSerializer(
                instance=keypair,
                data={
                    "name": keypair.name,
                    "certificate_data": keypair.certificate_data,
                    "key_data": keypair.key_data,
                },
            ).is_valid()
        )
        self.assertFalse(
            CertificateKeyPairSerializer(
                instance=keypair,
                data={
                    "name": keypair.name,
                    "certificate_data": "test",
                    "key_data": "test",
                },
            ).is_valid()
        )

    def test_builder(self):
        """Test Builder"""
        name = generate_id()
        builder = CertificateBuilder(name)
        with self.assertRaises(ValueError):
            builder.save()
        builder.build(
            subject_alt_names=[],
            validity_days=3,
        )
        instance = builder.save()
        _now = now()
        self.assertEqual(instance.name, name)
        self.assertEqual((instance.certificate.not_valid_after_utc - _now).days, 2)

    def test_builder_api(self):
        """Test Builder (via API)"""
        self.client.force_login(create_test_admin_user())
        name = generate_id()
        self.client.post(
            reverse("authentik_api:certificatekeypair-generate"),
            data={"common_name": name, "subject_alt_name": "bar,baz", "validity_days": 3},
        )
        key = CertificateKeyPair.objects.filter(name=name).first()
        self.assertIsNotNone(key)
        ext: SubjectAlternativeName = key.certificate.extensions[0].value
        self.assertIsInstance(ext, SubjectAlternativeName)
        self.assertIsInstance(ext[0], DNSName)
        self.assertEqual(ext[0].value, "bar")
        self.assertIsInstance(ext[1], DNSName)
        self.assertEqual(ext[1].value, "baz")

    def test_builder_api_duplicate(self):
        """Test Builder (via API)"""
        cert = create_test_cert()
        self.client.force_login(create_test_admin_user())
        res = self.client.post(
            reverse("authentik_api:certificatekeypair-generate"),
            data={"common_name": cert.name, "subject_alt_name": "bar,baz", "validity_days": 3},
        )
        self.assertEqual(res.status_code, 400)
        self.assertJSONEqual(res.content, {"common_name": ["This field must be unique."]})

    def test_builder_api_empty_san(self):
        """Test Builder (via API)"""
        self.client.force_login(create_test_admin_user())
        name = generate_id()
        self.client.post(
            reverse("authentik_api:certificatekeypair-generate"),
            data={"common_name": name, "subject_alt_name": "", "validity_days": 3},
        )
        key = CertificateKeyPair.objects.filter(name=name).first()
        self.assertIsNotNone(key)
        self.assertEqual(len(key.certificate.extensions), 0)

    def test_builder_api_empty_san_multiple(self):
        """Test Builder (via API)"""
        self.client.force_login(create_test_admin_user())
        name = generate_id()
        self.client.post(
            reverse("authentik_api:certificatekeypair-generate"),
            data={"common_name": name, "subject_alt_name": ", ", "validity_days": 3},
        )
        key = CertificateKeyPair.objects.filter(name=name).first()
        self.assertIsNotNone(key)
        self.assertEqual(len(key.certificate.extensions), 0)

    def test_builder_api_invalid(self):
        """Test Builder (via API) (invalid)"""
        self.client.force_login(create_test_admin_user())
        response = self.client.post(
            reverse("authentik_api:certificatekeypair-generate"),
            data={},
        )
        self.assertEqual(response.status_code, 400)

    def test_list(self):
        """Test API List"""
        cert = create_test_cert()
        self.client.force_login(create_test_admin_user())
        response = self.client.get(
            reverse(
                "authentik_api:certificatekeypair-list",
            ),
            data={"name": cert.name},
        )
        self.assertEqual(response.status_code, 200)
        body = loads(response.content.decode())
        api_cert = [x for x in body["results"] if x["name"] == cert.name][0]
        self.assertEqual(api_cert["fingerprint_sha1"], cert.fingerprint_sha1)
        self.assertEqual(api_cert["fingerprint_sha256"], cert.fingerprint_sha256)

    def test_list_has_key_false(self):
        """Test API List with has_key set to false"""
        cert = create_test_cert()
        cert.key_data = ""
        cert.save()
        self.client.force_login(create_test_admin_user())
        response = self.client.get(
            reverse(
                "authentik_api:certificatekeypair-list",
            ),
            data={"name": cert.name, "has_key": False},
        )
        self.assertEqual(response.status_code, 200)
        body = loads(response.content.decode())
        api_cert = [x for x in body["results"] if x["name"] == cert.name][0]
        self.assertEqual(api_cert["fingerprint_sha1"], cert.fingerprint_sha1)
        self.assertEqual(api_cert["fingerprint_sha256"], cert.fingerprint_sha256)

    def test_list_always_includes_details(self):
        """Test API List always includes certificate details"""
        cert = create_test_cert()
        self.client.force_login(create_test_admin_user())
        response = self.client.get(
            reverse(
                "authentik_api:certificatekeypair-list",
            ),
            data={"name": cert.name},
        )
        self.assertEqual(response.status_code, 200)
        body = loads(response.content.decode())
        api_cert = [x for x in body["results"] if x["name"] == cert.name][0]
        # All details should now always be included
        self.assertEqual(api_cert["fingerprint_sha1"], cert.fingerprint_sha1)
        self.assertEqual(api_cert["fingerprint_sha256"], cert.fingerprint_sha256)
        self.assertIsNotNone(api_cert["cert_expiry"])
        self.assertIsNotNone(api_cert["cert_subject"])

    def test_certificate_download(self):
        """Test certificate export (download)"""
        keypair = create_test_cert()
        user = create_test_user()
        user.assign_perms_to_managed_role("authentik_crypto.view_certificatekeypair", keypair)
        user.assign_perms_to_managed_role(
            "authentik_crypto.view_certificatekeypair_certificate", keypair
        )
        self.client.force_login(user)
        response = self.client.get(
            reverse(
                "authentik_api:certificatekeypair-view-certificate",
                kwargs={"pk": keypair.pk},
            )
        )
        self.assertEqual(response.status_code, 200)
        response = self.client.get(
            reverse(
                "authentik_api:certificatekeypair-view-certificate",
                kwargs={"pk": keypair.pk},
            ),
            data={"download": True},
        )
        self.assertEqual(response.status_code, 200)
        self.assertIn("Content-Disposition", response)

    def test_private_key_download(self):
        """Test private_key export (download)"""
        keypair = create_test_cert()
        user = create_test_user()
        user.assign_perms_to_managed_role("authentik_crypto.view_certificatekeypair", keypair)
        user.assign_perms_to_managed_role("authentik_crypto.view_certificatekeypair_key", keypair)
        self.client.force_login(user)
        response = self.client.get(
            reverse(
                "authentik_api:certificatekeypair-view-private-key",
                kwargs={"pk": keypair.pk},
            )
        )
        self.assertEqual(response.status_code, 200)
        response = self.client.get(
            reverse(
                "authentik_api:certificatekeypair-view-private-key",
                kwargs={"pk": keypair.pk},
            ),
            data={"download": True},
        )
        self.assertEqual(response.status_code, 200)
        self.assertIn("Content-Disposition", response)

    def test_certificate_download_denied(self):
        """Test certificate export (download)"""
        self.client.force_login(create_test_user())
        keypair = create_test_cert()
        response = self.client.get(
            reverse(
                "authentik_api:certificatekeypair-view-certificate",
                kwargs={"pk": keypair.pk},
            )
        )
        self.assertEqual(403, response.status_code)
        response = self.client.get(
            reverse(
                "authentik_api:certificatekeypair-view-certificate",
                kwargs={"pk": keypair.pk},
            ),
            data={"download": True},
        )
        self.assertEqual(403, response.status_code)

    def test_private_key_download_denied(self):
        """Test private_key export (download)"""
        self.client.force_login(create_test_user())
        keypair = create_test_cert()
        response = self.client.get(
            reverse(
                "authentik_api:certificatekeypair-view-private-key",
                kwargs={"pk": keypair.pk},
            )
        )
        self.assertEqual(403, response.status_code)
        response = self.client.get(
            reverse(
                "authentik_api:certificatekeypair-view-private-key",
                kwargs={"pk": keypair.pk},
            ),
            data={"download": True},
        )
        self.assertEqual(403, response.status_code)

    def test_used_by(self):
        """Test used_by endpoint"""
        self.client.force_login(create_test_admin_user())
        keypair = create_test_cert()
        provider = OAuth2Provider.objects.create(
            name=generate_id(),
            client_id=generate_id(),
            client_secret=generate_key(),
            authorization_flow=create_test_flow(),
            redirect_uris=[RedirectURI(RedirectURIMatchingMode.STRICT, "http://localhost")],
            signing_key=keypair,
        )
        response = self.client.get(
            reverse(
                "authentik_api:certificatekeypair-used-by",
                kwargs={"pk": keypair.pk},
            )
        )
        self.assertEqual(response.status_code, 200)
        self.assertJSONEqual(
            response.content.decode(),
            [
                {
                    "app": "authentik_providers_oauth2",
                    "model_name": "oauth2provider",
                    "pk": str(provider.pk),
                    "name": str(provider),
                    "action": DeleteAction.SET_NULL.value,
                }
            ],
        )

    def test_used_by_denied(self):
        """Test used_by endpoint"""
        self.client.logout()
        keypair = create_test_cert()
        OAuth2Provider.objects.create(
            name=generate_id(),
            client_id=generate_id(),
            client_secret=generate_key(),
            authorization_flow=create_test_flow(),
            redirect_uris=[RedirectURI(RedirectURIMatchingMode.STRICT, "http://localhost")],
            signing_key=keypair,
        )
        response = self.client.get(
            reverse(
                "authentik_api:certificatekeypair-used-by",
                kwargs={"pk": keypair.pk},
            )
        )
        self.assertEqual(403, response.status_code)

    def test_discovery(self):
        """Test certificate discovery"""
        # This test generates 2 separate cert/key combinations
        # and verifies they both import properly
        name = generate_id()
        builder = CertificateBuilder(name)
        with self.assertRaises(ValueError):
            builder.save()
        builder.build(
            subject_alt_names=[],
            validity_days=3,
        )

        name2 = generate_id()
        builder2 = CertificateBuilder(name2)
        with self.assertRaises(ValueError):
            builder2.save()
        builder2.build(
            subject_alt_names=[],
            validity_days=3,
        )

        name3 = generate_id()
        builder3 = CertificateBuilder(name3)
        with self.assertRaises(ValueError):
            builder3.save()
        builder3.build(
            subject_alt_names=[],
            validity_days=3,
        )

        with TemporaryDirectory() as temp_dir:
            with open(f"{temp_dir}/foo.pem", "w+", encoding="utf-8") as _cert:
                _cert.write(builder.certificate)
            with open(f"{temp_dir}/foo.key", "w+", encoding="utf-8") as _key:
                _key.write(builder.private_key)
            makedirs(f"{temp_dir}/foo.bar", exist_ok=True)
            with open(f"{temp_dir}/foo.bar/fullchain.pem", "w+", encoding="utf-8") as _cert:
                _cert.write(builder2.certificate)
            with open(f"{temp_dir}/foo.bar/privkey.pem", "w+", encoding="utf-8") as _key:
                _key.write(builder2.private_key)
            with open(f"{temp_dir}/tls-combined.pem", "w+", encoding="utf-8") as _cert:
                _cert.write(builder3.certificate)
            with CONFIG.patch("cert_discovery_dir", temp_dir):
                certificate_discovery.send()
        keypair: CertificateKeyPair = CertificateKeyPair.objects.filter(
            managed=MANAGED_DISCOVERED % "foo"
        ).first()
        self.assertIsNotNone(keypair)
        self.assertIsNotNone(keypair.certificate)
        self.assertIsNotNone(keypair.private_key)
        self.assertTrue(
            CertificateKeyPair.objects.filter(managed=MANAGED_DISCOVERED % "foo.bar").exists()
        )
        self.assertFalse(
            CertificateKeyPair.objects.filter(managed=MANAGED_DISCOVERED % "tls-combined").exists()
        )

    def test_discovery_updating_same_private_key(self):
        """Test certificate discovery updating certs with matching private keys"""
        name = generate_id()
        builder = CertificateBuilder(name)
        builder.build(
            subject_alt_names=[],
            validity_days=3,
        )

        with TemporaryDirectory() as temp_dir:
            # First discovery: write cert as "original"
            with open(f"{temp_dir}/original.pem", "w+", encoding="utf-8") as _cert:
                _cert.write(builder.certificate)
            with open(f"{temp_dir}/original.key", "w+", encoding="utf-8") as _key:
                _key.write(builder.private_key)

            with CONFIG.patch("cert_discovery_dir", temp_dir):
                certificate_discovery.send()

            # Verify "original" cert was created
            original = CertificateKeyPair.objects.filter(
                managed=MANAGED_DISCOVERED % "original"
            ).first()
            self.assertIsNotNone(original)
            self.assertEqual(original.name, "original")
            self.assertIsNotNone(original.private_key)

            # Second discovery: write same cert/key as "renamed"
            Path(f"{temp_dir}/original.pem").unlink()
            Path(f"{temp_dir}/original.key").unlink()

            with open(f"{temp_dir}/renamed.pem", "w+", encoding="utf-8") as _cert:
                _cert.write(builder.certificate)
            with open(f"{temp_dir}/renamed.key", "w+", encoding="utf-8") as _key:
                _key.write(builder.private_key)

            with CONFIG.patch("cert_discovery_dir", temp_dir):
                certificate_discovery.send()

            # Verify the cert was updated
            renamed = CertificateKeyPair.objects.filter(
                managed=MANAGED_DISCOVERED % "renamed"
            ).first()
            self.assertIsNotNone(renamed, "Renamed certificate should exist")
            self.assertEqual(renamed.name, "renamed")
            self.assertEqual(renamed.pk, original.pk, "Should be same database object")

            # Verify no new cert was created
            final_count = CertificateKeyPair.objects.filter(
                managed__startswith="goauthentik.io/crypto/discovered/"
            ).count()
            self.assertEqual(
                1, final_count, "Should not create duplicate cert for same private key"
            )

    def test_metadata_extraction_with_cert_and_key(self):
        """Test that metadata is extracted when creating keypair with certificate and key"""
        cert = create_test_cert()

        # Verify all metadata fields are populated
        self.assertIsNotNone(cert.key_type)
        self.assertIsNotNone(cert.cert_expiry)
        self.assertIsNotNone(cert.cert_subject)
        self.assertIsNotNone(cert.fingerprint_sha256)
        self.assertIsNotNone(cert.fingerprint_sha1)

        # Verify kid is generated using SHA512 for new records
        self.assertIsNotNone(cert.kid)
        self.assertEqual(cert.kid, generate_key_id(cert.key_data))

    def test_metadata_extraction_without_key(self):
        """Test that metadata is extracted when creating keypair without private key"""
        builder = CertificateBuilder(generate_id())
        builder.build(subject_alt_names=[], validity_days=3)

        # Create keypair with only certificate, no key
        cert = CertificateKeyPair.objects.create(
            name=generate_id(),
            certificate_data=builder.certificate,
            key_data="",
        )

        # Verify certificate metadata fields are populated
        self.assertIsNotNone(cert.key_type)
        self.assertIsNotNone(cert.cert_expiry)
        self.assertIsNotNone(cert.cert_subject)
        self.assertIsNotNone(cert.fingerprint_sha256)
        self.assertIsNotNone(cert.fingerprint_sha1)

        # Verify kid is empty when no key_data
        self.assertEqual(cert.kid, None)

    def test_metadata_extraction_invalid_cert(self):
        """Test that invalid certificate data doesn't crash, just skips metadata"""
        cert = CertificateKeyPair.objects.create(
            name=generate_id(),
            certificate_data="invalid certificate data",
            key_data="",
        )

        # Verify metadata fields are None for invalid cert
        self.assertIsNone(cert.key_type)
        self.assertIsNone(cert.cert_expiry)
        self.assertIsNone(cert.cert_subject)
        self.assertIsNone(cert.fingerprint_sha256)
        self.assertIsNone(cert.fingerprint_sha1)
        self.assertIsNone(cert.kid)

    def test_kid_legacy_preservation(self):
        """Test that legacy MD5 kid is preserved when key_data hasn't changed"""
        cert = create_test_cert()

        # Simulate a legacy MD5 kid (as if backfilled from old system)
        legacy_kid = generate_key_id_legacy(cert.key_data)
        CertificateKeyPair.objects.filter(pk=cert.pk).update(kid=legacy_kid)
        cert.refresh_from_db()
        self.assertEqual(cert.kid, legacy_kid)

        # Save the cert again (e.g., name change) - kid should be preserved
        cert.name = generate_id()
        cert.save()
        cert.refresh_from_db()

        self.assertEqual(cert.kid, legacy_kid)

    def test_kid_regenerated_on_key_change(self):
        """Test that kid is regenerated when key_data changes"""
        cert = create_test_cert()
        original_kid = cert.kid

        # Generate a new key and update the keypair
        builder = CertificateBuilder(generate_id())
        builder.build(subject_alt_names=[], validity_days=3)

        cert.key_data = builder.private_key
        cert.certificate_data = builder.certificate
        cert.save()
        cert.refresh_from_db()

        # Kid should be regenerated for the new key
        self.assertNotEqual(cert.kid, original_kid)
        self.assertEqual(cert.kid, generate_key_id(cert.key_data))

    def test_kid_regenerated_on_key_change_from_legacy(self):
        """Test that kid is regenerated from legacy MD5 when key_data changes"""
        cert = create_test_cert()

        # Simulate a legacy MD5 kid
        legacy_kid = generate_key_id_legacy(cert.key_data)
        CertificateKeyPair.objects.filter(pk=cert.pk).update(kid=legacy_kid)
        cert.refresh_from_db()
        self.assertEqual(cert.kid, legacy_kid)

        # Generate a new key and update the keypair
        builder = CertificateBuilder(generate_id())
        builder.build(subject_alt_names=[], validity_days=3)

        cert.key_data = builder.private_key
        cert.certificate_data = builder.certificate
        cert.save()
        cert.refresh_from_db()

        # Kid should now be SHA512 for the new key
        self.assertNotEqual(cert.kid, legacy_kid)
        self.assertEqual(cert.kid, generate_key_id(cert.key_data))
