# SPDX-FileCopyrightText: 2014 Legrandin <helderijs@gmail.com>
# SPDX-License-Identifier: BSD-2-Clause

from binascii import unhexlify

import pytest

from Crypto.Hash import SHA1, SHA3_224, SHA3_256, SHA3_384, SHA3_512, SHA224, SHA256, SHA384, SHA512
from Crypto.PublicKey import RSA
from Crypto.SelfTest.loader import load_test_vectors, load_test_vectors_wycheproof, wycheproof_id
from Crypto.SelfTest.st_common import wycheproof_warnings
from Crypto.Signature import PKCS1_v1_5, pkcs1_15
from Crypto.Util.number import bytes_to_long
from Crypto.Util.strxor import strxor


def load_hash_by_name(hash_name):
    return __import__("Crypto.Hash." + hash_name, globals(), locals(), ["new"])


class TestFIPS_PKCS1_Verify:
    def test_can_sign(self):
        test_public_key = RSA.generate(1024).public_key()
        verifier = pkcs1_15.new(test_public_key)
        assert verifier.can_sign() is False


test_vectors_verify = (
    load_test_vectors(
        ("Signature", "PKCS1-v1.5"),
        "SigVer15_186-3.rsp",
        "Signature Verification 186-3",
        {"shaalg": lambda x: x, "d": lambda x: int(x), "result": lambda x: x},
    )
    or []
)


def _load_verify_vectors():
    params = []
    for count, tv in enumerate(test_vectors_verify):
        if isinstance(tv, str):
            continue
        if hasattr(tv, "n"):
            modulus = tv.n
            continue

        hash_module = load_hash_by_name(tv.shaalg.upper())
        hash_obj = hash_module.new(tv.msg)
        public_key = RSA.construct([bytes_to_long(x) for x in (modulus, tv.e)])  # type: ignore
        verifier = pkcs1_15.new(public_key)
        params.append(pytest.param(hash_obj, verifier, tv.s, tv.result != "f", id=str(count)))
    return params


@pytest.mark.slow
class TestFIPS_PKCS1_Verify_Tests_KAT:
    @pytest.mark.parametrize("hash_obj, verifier, signature, valid", _load_verify_vectors())
    def test_verify(self, hash_obj, verifier, signature, valid):
        if valid:
            verifier.verify(hash_obj, signature)
        else:
            with pytest.raises(ValueError):
                verifier.verify(hash_obj, signature)


class TestFIPS_PKCS1_Sign:
    def test_can_sign(self):
        test_private_key = RSA.generate(1024)
        signer = pkcs1_15.new(test_private_key)
        assert signer.can_sign() is True


test_vectors_sign = (
    load_test_vectors(
        ("Signature", "PKCS1-v1.5"),
        "SigGen15_186-2.txt",
        "Signature Generation 186-2",
        {"shaalg": lambda x: x},
    )
    or []
)

test_vectors_sign += (
    load_test_vectors(
        ("Signature", "PKCS1-v1.5"),
        "SigGen15_186-3.txt",
        "Signature Generation 186-3",
        {"shaalg": lambda x: x},
    )
    or []
)


def _load_sign_vectors():
    params = []
    for count, tv in enumerate(test_vectors_sign):
        if isinstance(tv, str):
            continue
        if hasattr(tv, "n"):
            modulus = tv.n
            continue
        if hasattr(tv, "e"):
            private_key = RSA.construct([bytes_to_long(x) for x in (modulus, tv.e, tv.d)])  # type: ignore
            signer = pkcs1_15.new(private_key)
            continue

        hash_module = load_hash_by_name(tv.shaalg.upper())
        hash_obj = hash_module.new(tv.msg)
        params.append(pytest.param(hash_obj, signer, tv.s, id=str(count)))
    return params


@pytest.mark.slow
class TestFIPS_PKCS1_Sign_Tests_KAT:
    @pytest.mark.parametrize("hash_obj, signer, result", _load_sign_vectors())
    def test_sign(self, hash_obj, signer, result):
        signature = signer.sign(hash_obj)
        assert signature == result


class TestPKCS1_15_NoParams:
    """Verify that PKCS#1 v1.5 signatures pass even without NULL parameters in
    the algorithm identifier (PyCrypto/LP bug #1119552)."""

    rsakey = """-----BEGIN RSA PRIVATE KEY-----
            MIIBOwIBAAJBAL8eJ5AKoIsjURpcEoGubZMxLD7+kT+TLr7UkvEtFrRhDDKMtuII
            q19FrL4pUIMymPMSLBn3hJLe30Dw48GQM4UCAwEAAQJACUSDEp8RTe32ftq8IwG8
            Wojl5mAd1wFiIOrZ/Uv8b963WJOJiuQcVN29vxU5+My9GPZ7RA3hrDBEAoHUDPrI
            OQIhAPIPLz4dphiD9imAkivY31Rc5AfHJiQRA7XixTcjEkojAiEAyh/pJHks/Mlr
            +rdPNEpotBjfV4M4BkgGAA/ipcmaAjcCIQCHvhwwKVBLzzTscT2HeUdEeBMoiXXK
            JACAr3sJQJGxIQIgarRp+m1WSKV1MciwMaTOnbU7wxFs9DP1pva76lYBzgUCIQC9
            n0CnZCJ6IZYqSt0H5N7+Q+2Ro64nuwV/OSQfM6sBwQ==
            -----END RSA PRIVATE KEY-----"""

    msg = b"This is a test\x0a"

    # PKCS1 v1.5 signature of the message computed using SHA-1.
    # The digestAlgorithm SEQUENCE does NOT contain the NULL parameter.
    sig_str = (
        "a287a13517f716e72fb14eea8e33a8db4a4643314607e7ca3e3e28"
        "1893db74013dda8b855fd99f6fecedcb25fcb7a434f35cd0a101f8"
        "b19348e0bd7b6f152dfc"
    )
    signature = unhexlify(sig_str)

    def test(self):
        verifier = pkcs1_15.new(RSA.importKey(self.rsakey))
        hashed = SHA1.new(self.msg)
        verifier.verify(hashed, self.signature)


class TestPKCS1_Legacy_Module:
    """Verify that the legacy module Crypto.Signature.PKCS1_v1_5
    behaves as expected. The only difference is that the verify()
    method returns True/False and does not raise exceptions."""

    def test(self):
        key = RSA.importKey(TestPKCS1_15_NoParams.rsakey)
        hashed = SHA1.new(b"Test")
        good_signature = PKCS1_v1_5.new(key).sign(hashed)
        verifier = PKCS1_v1_5.new(key.public_key())

        assert verifier.verify(hashed, good_signature) is True

        # Flip a few bits in the signature
        bad_signature = strxor(good_signature, bytes([1]) * len(good_signature))
        assert verifier.verify(hashed, bad_signature) is False


class TestPKCS1_All_Hashes:
    def test(self):
        key = RSA.generate(1024)
        signer = pkcs1_15.new(key)
        hash_names = (
            "MD2",
            "MD4",
            "MD5",
            "RIPEMD160",
            "SHA1",
            "SHA224",
            "SHA256",
            "SHA384",
            "SHA512",
            "SHA3_224",
            "SHA3_256",
            "SHA3_384",
            "SHA3_512",
        )

        for name in hash_names:
            hashed = load_hash_by_name(name).new(b"Test")
            signer.sign(hashed)

        from Crypto.Hash import BLAKE2b, BLAKE2s

        for hash_size in (20, 32, 48, 64):
            hashed_b = BLAKE2b.new(digest_bytes=hash_size, data=b"Test")
            signer.sign(hashed_b)
        for hash_size in (16, 20, 28, 32):
            hashed_s = BLAKE2s.new(digest_bytes=hash_size, data=b"Test")
            signer.sign(hashed_s)


def _load_tests(filename):
    def filter_rsa(group):
        # Verification tests have the public key, generation tests the private key too
        if "privateKeyPem" in group:
            return RSA.import_key(group["privateKeyPem"])
        return RSA.import_key(group["publicKeyPem"])

    def filter_sha(group):
        hash_name = group["sha"]
        if hash_name == "SHA-512":
            return SHA512
        elif hash_name == "SHA-512/224":
            return SHA512.new(truncate="224")
        elif hash_name == "SHA-512/256":
            return SHA512.new(truncate="256")
        elif hash_name == "SHA3-512":
            return SHA3_512
        elif hash_name == "SHA-384":
            return SHA384
        elif hash_name == "SHA3-384":
            return SHA3_384
        elif hash_name == "SHA-256":
            return SHA256
        elif hash_name == "SHA3-256":
            return SHA3_256
        elif hash_name == "SHA-224":
            return SHA224
        elif hash_name == "SHA3-224":
            return SHA3_224
        elif hash_name == "SHA-1":
            return SHA1
        else:
            raise ValueError("Unknown hash algorithm: " + hash_name)

    def filter_type(group):
        type_name = group["type"]
        if type_name not in ("RsassaPkcs1Verify", "RsassaPkcs1Generate"):
            raise ValueError("Unknown type name " + type_name)
        return type_name

    result = load_test_vectors_wycheproof(
        ("Signature", "wycheproof"),
        filename,
        "Wycheproof PKCS#1v1.5 signature (%s)" % filename,
        group_tag={"rsa_key": filter_rsa, "hash_mod": filter_sha, "type": filter_type},
    )
    return result


def load_wycheproof_vectors():
    vectors = []
    for bits in (1024, 1536, 2048, 3072, 4096):
        vectors += _load_tests("rsa_pkcs1_%d_sig_gen_test.json" % bits)
    vectors += _load_tests("rsa_signature_2048_sha224_test.json")
    vectors += _load_tests("rsa_signature_2048_sha256_test.json")
    vectors += _load_tests("rsa_signature_2048_sha384_test.json")
    vectors += _load_tests("rsa_signature_2048_sha3_224_test.json")
    vectors += _load_tests("rsa_signature_2048_sha3_256_test.json")
    vectors += _load_tests("rsa_signature_2048_sha3_384_test.json")
    vectors += _load_tests("rsa_signature_2048_sha3_512_test.json")
    vectors += _load_tests("rsa_signature_2048_sha512_test.json")
    vectors += _load_tests("rsa_signature_2048_sha512_224_test.json")
    vectors += _load_tests("rsa_signature_2048_sha512_256_test.json")
    vectors += _load_tests("rsa_signature_3072_sha256_test.json")
    vectors += _load_tests("rsa_signature_3072_sha384_test.json")
    vectors += _load_tests("rsa_signature_3072_sha3_256_test.json")
    vectors += _load_tests("rsa_signature_3072_sha3_384_test.json")
    vectors += _load_tests("rsa_signature_3072_sha3_512_test.json")
    vectors += _load_tests("rsa_signature_3072_sha512_test.json")
    vectors += _load_tests("rsa_signature_3072_sha512_256_test.json")
    vectors += _load_tests("rsa_signature_4096_sha256_test.json")
    vectors += _load_tests("rsa_signature_4096_sha384_test.json")
    vectors += _load_tests("rsa_signature_4096_sha512_test.json")
    vectors += _load_tests("rsa_signature_4096_sha512_256_test.json")
    vectors += _load_tests("rsa_signature_8192_sha256_test.json")
    vectors += _load_tests("rsa_signature_8192_sha384_test.json")
    vectors += _load_tests("rsa_signature_8192_sha512_test.json")
    return vectors


class TestVectorsWycheproof:
    _id = "None"

    def warn(self, tv):
        if tv.warning and wycheproof_warnings():
            import warnings

            warnings.warn("Wycheproof warning: %s (%s)" % (self._id, tv.comment))

    def check_sign(self, tv):
        # The signature is deterministic
        if tv.type != "RsassaPkcs1Generate" or not tv.valid:
            return
        signature = pkcs1_15.new(tv.rsa_key).sign(tv.hash_mod.new(tv.msg))
        assert signature == tv.sig

    def check_verify(self, tv):
        self._id = "Wycheproof RSA PKCS$#1 Test #" + str(tv.id)

        hashed_msg = tv.hash_mod.new(tv.msg)
        signer = pkcs1_15.new(tv.rsa_key.public_key())
        try:
            signer.verify(hashed_msg, tv.sig)
        except ValueError:
            if tv.warning:
                return
            assert not tv.valid
        else:
            assert tv.valid
            self.warn(tv)

    @pytest.mark.parametrize("tv", load_wycheproof_vectors(), ids=wycheproof_id)
    def test(self, tv):
        self.check_sign(tv)
        self.check_verify(tv)
