broker/tests/test_hasher.py

import base64
import hashlib
import os
import sys
import unittest
from unittest import mock

sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), os.pardir)))

import coreci_broker

SCRYPT_N = 2 ** 14
SCRYPT_R = 8
SCRYPT_P = 1

BCRYPT_SHAPED = "$2b$12$K7L9O1P3QRSTUVWXYZabcdefghijk012345678901234567890123456789"


def _scrypt_stored(token, salt):
    digest = hashlib.scrypt(
        token.encode("utf-8"), salt=salt, n=SCRYPT_N, r=SCRYPT_R, p=SCRYPT_P
    )
    return "{}${}${}".format(
        "scrypt",
        base64.urlsafe_b64encode(salt).decode("ascii"),
        base64.urlsafe_b64encode(digest).decode("ascii"),
    )


class _FakeBcryptModule:
    def __init__(self):
        self.gensalt_calls = []
        self.hashpw_calls = []
        self.checkpw_calls = []

    def gensalt(self):
        self.gensalt_calls.append(True)
        return b"$2b$12$fakesaltfakesalt"

    def hashpw(self, password, salt):
        self.hashpw_calls.append((password, salt))
        return b"$2b$12$storedhashvalue"

    def checkpw(self, password, hashed):
        self.checkpw_calls.append((password, hashed))
        return password == b"mock-token"


class HashFormatTests(unittest.TestCase):
    def test_bcrypt_is_not_importable_on_this_host(self):
        self.assertIsNone(coreci_broker.bcrypt)
        with self.assertRaises(ImportError):
            import bcrypt

    def test_hash_output_starts_with_scrypt_prefix(self):
        stored = coreci_broker.hash_token("format-token")
        self.assertTrue(stored.startswith("scrypt$"))

    def test_hash_output_carries_salt_and_digest_b64url_fields(self):
        stored = coreci_broker.hash_token("format-token")
        fields = stored.split("$")
        self.assertEqual(3, len(fields))
        self.assertEqual("scrypt", fields[0])
        salt = base64.urlsafe_b64decode(fields[1].encode("ascii"))
        digest = base64.urlsafe_b64decode(fields[2].encode("ascii"))
        self.assertEqual(16, len(salt))
        self.assertEqual(64, len(digest))


class RoundtripTests(unittest.TestCase):
    def test_hash_then_verify_roundtrip(self):
        stored = coreci_broker.hash_token("roundtrip-token")
        self.assertTrue(coreci_broker.verify_token("roundtrip-token", stored))

    def test_wrong_token_fails(self):
        stored = coreci_broker.hash_token("roundtrip-token")
        self.assertFalse(coreci_broker.verify_token("wrong-token", stored))

    def test_two_hashes_of_same_token_differ(self):
        first = coreci_broker.hash_token("same-token")
        second = coreci_broker.hash_token("same-token")
        self.assertNotEqual(first, second)
        self.assertTrue(coreci_broker.verify_token("same-token", first))
        self.assertTrue(coreci_broker.verify_token("same-token", second))


class ScryptParseTests(unittest.TestCase):
    def test_hand_built_stored_value_verifies(self):
        stored = _scrypt_stored("parse-token", bytes(range(16)))
        self.assertTrue(stored.startswith("scrypt$"))
        self.assertTrue(coreci_broker.verify_token("parse-token", stored))

    def test_hand_built_stored_value_rejects_other_token(self):
        stored = _scrypt_stored("parse-token", bytes(range(16)))
        self.assertFalse(coreci_broker.verify_token("parse-token-x", stored))

    def test_tampered_digest_field_fails(self):
        salt = bytes(range(16))
        digest = hashlib.scrypt(
            b"parse-token", salt=salt, n=SCRYPT_N, r=SCRYPT_R, p=SCRYPT_P
        )
        tampered = digest[:-1] + bytes([digest[-1] ^ 0x01])
        stored = "{}${}${}".format(
            "scrypt",
            base64.urlsafe_b64encode(salt).decode("ascii"),
            base64.urlsafe_b64encode(tampered).decode("ascii"),
        )
        self.assertFalse(coreci_broker.verify_token("parse-token", stored))

    def test_malformed_scrypt_values_fail_cleanly(self):
        for stored in (
            "scrypt$",
            "scrypt$abc",
            "scrypt$abc$def$ghi",
            "scrypt$$",
            "scrypt$!@#$%",
        ):
            with self.subTest(stored=stored):
                self.assertFalse(coreci_broker.verify_token("x", stored))


class UnknownPrefixTests(unittest.TestCase):
    def test_unknown_prefixes_return_false_without_raising(self):
        for stored in (
            "",
            "plaintext",
            "$2c$12$abc",
            "$2x$12$abc",
            "argon2$abc$def",
            "SCRYPT$abc$def",
            "scrypt2$abc$def",
            "0123456789",
        ):
            with self.subTest(stored=stored):
                self.assertFalse(coreci_broker.verify_token("x", stored))

    def test_non_string_stored_returns_false_cleanly(self):
        self.assertFalse(coreci_broker.verify_token("x", None))
        self.assertFalse(coreci_broker.verify_token("x", 5))
        self.assertFalse(coreci_broker.verify_token("x", ["scrypt$abc$def"]))


class BcryptDispatchContractTests(unittest.TestCase):
    def test_bcrypt_prefix_returns_false_cleanly_with_bcrypt_absent(self):
        self.assertIsNone(coreci_broker.bcrypt)
        with self.assertRaises(ImportError):
            import bcrypt
        self.assertFalse(coreci_broker.verify_token("x", BCRYPT_SHAPED))

    def test_all_bcrypt_prefixes_return_false_cleanly_with_bcrypt_absent(self):
        self.assertIsNone(coreci_broker.bcrypt)
        for prefix in ("$2a$", "$2b$", "$2y$"):
            with self.subTest(prefix=prefix):
                self.assertFalse(coreci_broker.verify_token("x", prefix + "12$storedhashvalue"))


class BcryptBranchTests(unittest.TestCase):
    def test_hash_token_uses_bcrypt_branch_when_importable(self):
        fake = _FakeBcryptModule()
        with mock.patch.object(coreci_broker, "bcrypt", fake):
            stored = coreci_broker.hash_token("mock-token")
        self.assertEqual("$2b$12$storedhashvalue", stored)
        self.assertEqual([True], fake.gensalt_calls)
        self.assertEqual([(b"mock-token", b"$2b$12$fakesaltfakesalt")], fake.hashpw_calls)

    def test_verify_dispatches_bcrypt_prefixes_to_bcrypt_branch(self):
        fake = _FakeBcryptModule()
        expected = [
            (b"mock-token", (prefix + "12$storedhashvalue").encode("ascii"))
            for prefix in ("$2a$", "$2b$", "$2y$")
        ]
        with mock.patch.object(coreci_broker, "bcrypt", fake):
            for password, hashed in expected:
                with self.subTest(hashed=hashed):
                    self.assertTrue(coreci_broker.verify_token(password.decode("utf-8"), hashed.decode("ascii")))
        self.assertEqual(expected, fake.checkpw_calls)

    def test_bcrypt_checkpw_rejection_returns_false(self):
        fake = _FakeBcryptModule()
        with mock.patch.object(coreci_broker, "bcrypt", fake):
            self.assertFalse(coreci_broker.verify_token("other-token", "$2b$12$storedhashvalue"))
        self.assertEqual([(b"other-token", b"$2b$12$storedhashvalue")], fake.checkpw_calls)

    def test_bcrypt_checkpw_error_returns_false_cleanly(self):
        fake = _FakeBcryptModule()

        def broken(password, hashed):
            raise ValueError("invalid salt")

        fake.checkpw = broken
        with mock.patch.object(coreci_broker, "bcrypt", fake):
            self.assertFalse(coreci_broker.verify_token("mock-token", "$2b$garbage"))

    def test_scrypt_and_unknown_stored_never_dispatch_to_bcrypt(self):
        fake = _FakeBcryptModule()
        stored = coreci_broker.hash_token("mock-token")
        self.assertTrue(stored.startswith("scrypt$"))
        with mock.patch.object(coreci_broker, "bcrypt", fake):
            self.assertTrue(coreci_broker.verify_token("mock-token", stored))
            self.assertFalse(coreci_broker.verify_token("other", stored))
            self.assertFalse(coreci_broker.verify_token("mock-token", "unknown$prefix"))
        self.assertEqual([], fake.checkpw_calls)
        self.assertEqual([], fake.hashpw_calls)


if __name__ == "__main__":
    unittest.main()