from __future__ import with_statement, division

try:
    import unittest2 as unittest
except ImportError:
    import unittest
import os
import shutil
import subprocess
import pytest
import sys
from binascii import hexlify, unhexlify
import hashlib
from functools import partial

from hypothesis import given, settings
import hypothesis.strategies as st

from six import b, print_, binary_type
from .keys import SigningKey, VerifyingKey
from .keys import BadSignatureError, MalformedPointError, BadDigestError
from . import util
from .util import (
    sigencode_der,
    sigencode_strings,
    sigencode_strings_canonize,
    sigencode_string_canonize,
    sigencode_der_canonize,
)
from .util import sigdecode_der, sigdecode_strings, sigdecode_string
from .util import number_to_string, encoded_oid_ecPublicKey, MalformedSignature
from .curves import Curve, UnknownCurveError
from .curves import (
    SECP112r1,
    SECP112r2,
    SECP128r1,
    SECP160r1,
    NIST192p,
    NIST224p,
    NIST256p,
    NIST384p,
    NIST521p,
    SECP256k1,
    BRAINPOOLP160r1,
    BRAINPOOLP192r1,
    BRAINPOOLP224r1,
    BRAINPOOLP256r1,
    BRAINPOOLP320r1,
    BRAINPOOLP384r1,
    BRAINPOOLP512r1,
    BRAINPOOLP160t1,
    BRAINPOOLP192t1,
    BRAINPOOLP224t1,
    BRAINPOOLP256t1,
    BRAINPOOLP320t1,
    BRAINPOOLP384t1,
    BRAINPOOLP512t1,
    Ed25519,
    Ed448,
    curves,
)
from .ecdsa import (
    curve_brainpoolp224r1,
    curve_brainpoolp256r1,
    curve_brainpoolp384r1,
    curve_brainpoolp512r1,
)
from .ellipticcurve import Point
from . import der
from . import rfc6979
from . import ecdsa


class SubprocessError(Exception):
    pass


HYP_SETTINGS = {}


if "--fast" in sys.argv:  # pragma: no cover
    HYP_SETTINGS["max_examples"] = 2


def run_openssl(cmd):
    OPENSSL = "openssl"
    p = subprocess.Popen(
        [OPENSSL] + cmd.split(),
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
    )
    stdout, ignored = p.communicate()
    if p.returncode != 0:
        raise SubprocessError(
            "cmd '%s %s' failed: rc=%s, stdout/err was %s"
            % (OPENSSL, cmd, p.returncode, stdout)
        )
    return stdout.decode()


class ECDSA(unittest.TestCase):
    def test_basic(self):
        priv = SigningKey.generate()
        pub = priv.get_verifying_key()

        data = b"blahblah"
        sig = priv.sign(data)

        self.assertTrue(pub.verify(sig, data))
        self.assertRaises(BadSignatureError, pub.verify, sig, data + b"bad")

        pub2 = VerifyingKey.from_string(pub.to_string())
        self.assertTrue(pub2.verify(sig, data))

    def test_deterministic(self):
        data = b"blahblah"
        secexp = int("9d0219792467d7d37b4d43298a7d0c05", 16)

        priv = SigningKey.from_secret_exponent(
            secexp, SECP256k1, hashlib.sha256
        )
        pub = priv.get_verifying_key()

        k = rfc6979.generate_k(
            SECP256k1.generator.order(),
            secexp,
            hashlib.sha256,
            hashlib.sha256(data).digest(),
        )

        sig1 = priv.sign(data, k=k)
        self.assertTrue(pub.verify(sig1, data))

        sig2 = priv.sign(data, k=k)
        self.assertTrue(pub.verify(sig2, data))

        sig3 = priv.sign_deterministic(data, hashlib.sha256)
        self.assertTrue(pub.verify(sig3, data))

        self.assertEqual(sig1, sig2)
        self.assertEqual(sig1, sig3)

    def test_bad_usage(self):
        # sk=SigningKey() is wrong
        self.assertRaises(TypeError, SigningKey)
        self.assertRaises(TypeError, VerifyingKey)

    def test_lengths_default(self):
        default = NIST192p
        priv = SigningKey.generate()
        pub = priv.get_verifying_key()
        self.assertEqual(len(pub.to_string()), default.verifying_key_length)
        sig = priv.sign(b"data")
        self.assertEqual(len(sig), default.signature_length)

    def test_serialize(self):
        seed = b"secret"
        curve = NIST192p
        secexp1 = util.randrange_from_seed__trytryagain(seed, curve.order)
        secexp2 = util.randrange_from_seed__trytryagain(seed, curve.order)
        self.assertEqual(secexp1, secexp2)
        priv1 = SigningKey.from_secret_exponent(secexp1, curve)
        priv2 = SigningKey.from_secret_exponent(secexp2, curve)
        self.assertEqual(
            hexlify(priv1.to_string()), hexlify(priv2.to_string())
        )
        self.assertEqual(priv1.to_pem(), priv2.to_pem())
        pub1 = priv1.get_verifying_key()
        pub2 = priv2.get_verifying_key()
        data = b"data"
        sig1 = priv1.sign(data)
        sig2 = priv2.sign(data)
        self.assertTrue(pub1.verify(sig1, data))
        self.assertTrue(pub2.verify(sig1, data))
        self.assertTrue(pub1.verify(sig2, data))
        self.assertTrue(pub2.verify(sig2, data))
        self.assertEqual(hexlify(pub1.to_string()), hexlify(pub2.to_string()))

    def test_nonrandom(self):
        s = b"all the entropy in the entire world, compressed into one line"

        def not_much_entropy(numbytes):
            return s[:numbytes]

        # we control the entropy source, these two keys should be identical:
        priv1 = SigningKey.generate(entropy=not_much_entropy)
        priv2 = SigningKey.generate(entropy=not_much_entropy)
        self.assertEqual(
            hexlify(priv1.get_verifying_key().to_string()),
            hexlify(priv2.get_verifying_key().to_string()),
        )
        # likewise, signatures should be identical. Obviously you'd never
        # want to do this with keys you care about, because the secrecy of
        # the private key depends upon using different random numbers for
        # each signature
        sig1 = priv1.sign(b"data", entropy=not_much_entropy)
        sig2 = priv2.sign(b"data", entropy=not_much_entropy)
        self.assertEqual(hexlify(sig1), hexlify(sig2))

    def assertTruePrivkeysEqual(self, priv1, priv2):
        self.assertEqual(
            priv1.privkey.secret_multiplier, priv2.privkey.secret_multiplier
        )
        self.assertEqual(
            priv1.privkey.public_key.generator,
            priv2.privkey.public_key.generator,
        )

    def test_privkey_creation(self):
        s = b"all the entropy in the entire world, compressed into one line"

        def not_much_entropy(numbytes):
            return s[:numbytes]

        priv1 = SigningKey.generate()
        self.assertEqual(priv1.baselen, NIST192p.baselen)

        priv1 = SigningKey.generate(curve=NIST224p)
        self.assertEqual(priv1.baselen, NIST224p.baselen)

        priv1 = SigningKey.generate(entropy=not_much_entropy)
        self.assertEqual(priv1.baselen, NIST192p.baselen)
        priv2 = SigningKey.generate(entropy=not_much_entropy)
        self.assertEqual(priv2.baselen, NIST192p.baselen)
        self.assertTruePrivkeysEqual(priv1, priv2)

        priv1 = SigningKey.from_secret_exponent(secexp=3)
        self.assertEqual(priv1.baselen, NIST192p.baselen)
        priv2 = SigningKey.from_secret_exponent(secexp=3)
        self.assertTruePrivkeysEqual(priv1, priv2)

        priv1 = SigningKey.from_secret_exponent(secexp=4, curve=NIST224p)
        self.assertEqual(priv1.baselen, NIST224p.baselen)

    def test_privkey_strings(self):
        priv1 = SigningKey.generate()
        s1 = priv1.to_string()
        self.assertEqual(type(s1), binary_type)
        self.assertEqual(len(s1), NIST192p.baselen)
        priv2 = SigningKey.from_string(s1)
        self.assertTruePrivkeysEqual(priv1, priv2)

        s1 = priv1.to_pem()
        self.assertEqual(type(s1), binary_type)
        self.assertTrue(s1.startswith(b"-----BEGIN EC PRIVATE KEY-----"))
        self.assertTrue(s1.strip().endswith(b"-----END EC PRIVATE KEY-----"))
        priv2 = SigningKey.from_pem(s1)
        self.assertTruePrivkeysEqual(priv1, priv2)

        s1 = priv1.to_der()
        self.assertEqual(type(s1), binary_type)
        priv2 = SigningKey.from_der(s1)
        self.assertTruePrivkeysEqual(priv1, priv2)

        priv1 = SigningKey.generate(curve=NIST256p)
        s1 = priv1.to_pem()
        self.assertEqual(type(s1), binary_type)
        self.assertTrue(s1.startswith(b"-----BEGIN EC PRIVATE KEY-----"))
        self.assertTrue(s1.strip().endswith(b"-----END EC PRIVATE KEY-----"))
        priv2 = SigningKey.from_pem(s1)
        self.assertTruePrivkeysEqual(priv1, priv2)

        s1 = priv1.to_der()
        self.assertEqual(type(s1), binary_type)
        priv2 = SigningKey.from_der(s1)
        self.assertTruePrivkeysEqual(priv1, priv2)

    def test_privkey_strings_brainpool(self):
        priv1 = SigningKey.generate(curve=BRAINPOOLP512r1)
        s1 = priv1.to_pem()
        self.assertEqual(type(s1), binary_type)
        self.assertTrue(s1.startswith(b"-----BEGIN EC PRIVATE KEY-----"))
        self.assertTrue(s1.strip().endswith(b"-----END EC PRIVATE KEY-----"))
        priv2 = SigningKey.from_pem(s1)
        self.assertTruePrivkeysEqual(priv1, priv2)

        s1 = priv1.to_der()
        self.assertEqual(type(s1), binary_type)
        priv2 = SigningKey.from_der(s1)
        self.assertTruePrivkeysEqual(priv1, priv2)

    def assertTruePubkeysEqual(self, pub1, pub2):
        self.assertEqual(pub1.pubkey.point, pub2.pubkey.point)
        self.assertEqual(pub1.pubkey.generator, pub2.pubkey.generator)
        self.assertEqual(pub1.curve, pub2.curve)

    def test_pubkey_strings(self):
        priv1 = SigningKey.generate()
        pub1 = priv1.get_verifying_key()
        s1 = pub1.to_string()
        self.assertEqual(type(s1), binary_type)
        self.assertEqual(len(s1), NIST192p.verifying_key_length)
        pub2 = VerifyingKey.from_string(s1)
        self.assertTruePubkeysEqual(pub1, pub2)

        priv1 = SigningKey.generate(curve=NIST256p)
        pub1 = priv1.get_verifying_key()
        s1 = pub1.to_string()
        self.assertEqual(type(s1), binary_type)
        self.assertEqual(len(s1), NIST256p.verifying_key_length)
        pub2 = VerifyingKey.from_string(s1, curve=NIST256p)
        self.assertTruePubkeysEqual(pub1, pub2)

        pub1_der = pub1.to_der()
        self.assertEqual(type(pub1_der), binary_type)
        pub2 = VerifyingKey.from_der(pub1_der)
        self.assertTruePubkeysEqual(pub1, pub2)

        self.assertRaises(
            der.UnexpectedDER, VerifyingKey.from_der, pub1_der + b"junk"
        )
        badpub = VerifyingKey.from_der(pub1_der)

        class FakeGenerator:
            def order(self):
                return 123456789

        class FakeCurveFp:
            def p(self):
                return int(
                    "6525534529039240705020950546962731340"
                    "4541085228058844382513856749047873406763"
                )

        badcurve = Curve(
            "unknown", FakeCurveFp(), FakeGenerator(), (1, 2, 3, 4, 5, 6), None
        )
        badpub.curve = badcurve
        badder = badpub.to_der()
        self.assertRaises(UnknownCurveError, VerifyingKey.from_der, badder)

        pem = pub1.to_pem()
        self.assertEqual(type(pem), binary_type)
        self.assertTrue(pem.startswith(b"-----BEGIN PUBLIC KEY-----"), pem)
        self.assertTrue(pem.strip().endswith(b"-----END PUBLIC KEY-----"), pem)
        pub2 = VerifyingKey.from_pem(pem)
        self.assertTruePubkeysEqual(pub1, pub2)

    def test_pubkey_strings_brainpool(self):
        priv1 = SigningKey.generate(curve=BRAINPOOLP512r1)
        pub1 = priv1.get_verifying_key()
        s1 = pub1.to_string()
        self.assertEqual(type(s1), binary_type)
        self.assertEqual(len(s1), BRAINPOOLP512r1.verifying_key_length)
        pub2 = VerifyingKey.from_string(s1, curve=BRAINPOOLP512r1)
        self.assertTruePubkeysEqual(pub1, pub2)

        pub1_der = pub1.to_der()
        self.assertEqual(type(pub1_der), binary_type)
        pub2 = VerifyingKey.from_der(pub1_der)
        self.assertTruePubkeysEqual(pub1, pub2)

    def test_vk_to_der_with_invalid_point_encoding(self):
        sk = SigningKey.generate()
        vk = sk.verifying_key

        with self.assertRaises(ValueError):
            vk.to_der("raw")

    def test_sk_to_der_with_invalid_point_encoding(self):
        sk = SigningKey.generate()

        with self.assertRaises(ValueError):
            sk.to_der("raw")

    def test_vk_from_der_garbage_after_curve_oid(self):
        type_oid_der = encoded_oid_ecPublicKey
        curve_oid_der = der.encode_oid(*(1, 2, 840, 10045, 3, 1, 1)) + b(
            "garbage"
        )
        enc_type_der = der.encode_sequence(type_oid_der, curve_oid_der)
        point_der = der.encode_bitstring(b"\x00\xff", None)
        to_decode = der.encode_sequence(enc_type_der, point_der)

        with self.assertRaises(der.UnexpectedDER):
            VerifyingKey.from_der(to_decode)

    def test_vk_from_der_invalid_key_type(self):
        type_oid_der = der.encode_oid(*(1, 2, 3))
        curve_oid_der = der.encode_oid(*(1, 2, 840, 10045, 3, 1, 1))
        enc_type_der = der.encode_sequence(type_oid_der, curve_oid_der)
        point_der = der.encode_bitstring(b"\x00\xff", None)
        to_decode = der.encode_sequence(enc_type_der, point_der)

        with self.assertRaises(der.UnexpectedDER):
            VerifyingKey.from_der(to_decode)

    def test_vk_from_der_garbage_afte