From 897b981098133043a1858cc271b73edb2099891d Mon Sep 17 00:00:00 2001 From: Donovan Date: Mon, 15 Jul 2024 13:01:28 -0500 Subject: [PATCH] implement generate login interface and login --- src/nkode_interface.py | 2 +- src/pseudo_nkode_api.py | 82 ++++++++++++++++++++++++++-------- src/user_cipher_keys.py | 84 +++++++++++++++++++++++++---------- src/user_interface.py | 27 +++++++++-- src/utils.py | 27 +++-------- test/test_encipher.py | 5 --- test/test_pseudo_nkode_api.py | 5 +++ test/test_user_cipher_keys.py | 35 +++++++++++++++ test/test_user_interface.py | 30 ++++++++++--- 9 files changed, 220 insertions(+), 77 deletions(-) delete mode 100644 test/test_encipher.py create mode 100644 test/test_user_cipher_keys.py diff --git a/src/nkode_interface.py b/src/nkode_interface.py index 4f5e5f9..a6eb84a 100644 --- a/src/nkode_interface.py +++ b/src/nkode_interface.py @@ -20,7 +20,7 @@ class CustomerInterface(BaseModel): customer_interface=generate_random_nonrepeating_list(attrs_per_key*numb_keys), set_vals=generate_random_nonrepeating_list(attrs_per_key), numb_keys=numb_keys, - attrs_per_key=attrs_per_key, + attrs_per_key=attrs_per_key, ) def get_interface_by_set(self) -> dict[int, list[int]]: diff --git a/src/pseudo_nkode_api.py b/src/pseudo_nkode_api.py index 792e9db..84905b3 100644 --- a/src/pseudo_nkode_api.py +++ b/src/pseudo_nkode_api.py @@ -13,12 +13,43 @@ class UserDBModel(BaseModel): user_keys: UserCipherKeys user_interface: UserInterface + def get_passcode_set_vals(self) -> list[int]: + pass + class CustomerDBModel(BaseModel): customer_id: UUID interface: CustomerInterface - users: list[UserDBModel] + users: dict[str, UserDBModel] + def add_new_user(self, user: UserDBModel): + self.users[user.username] = user + + def valid_key_entry(self, username, selected_keys) -> bool: + assert (username in self.users.keys()) + assert (all(key_idx < self.interface.numb_keys for key_idx in selected_keys)) + passcode_len = len(selected_keys) + user = self.users[username] + + passcode_set_vals = user.user_keys.decipher_mask( + user.enciphered_passcode.mask, self.interface.set_vals, len(selected_keys)) + + set_vals_idx = [self.interface.get_set_index(set_val) for set_val in passcode_set_vals] + + presumed_selected_attributes = [] + for idx in range(passcode_len): + key_numb = selected_keys[idx] + key_attribute_idxs = user.user_interface.get_key_attr_idxs(key_numb) + + set_idx = set_vals_idx[idx] + selected_attr_idx = key_attribute_idxs[set_idx] + presumed_selected_attributes.append(selected_attr_idx) + + enciphered_attr = user.user_keys.encipher_salt_hash_code(presumed_selected_attributes, self.interface) + if enciphered_attr == user.enciphered_passcode.code: + return True + + return False class SessionCacheModel(BaseModel): session_id: UUID @@ -48,19 +79,29 @@ class PseudoNKodeAPI(BaseModel): self.sessions[new_session.session_id] = new_session return new_session.session_id, new_session.set_interface - def get_login_index_interface(self): - pass + def get_login_index_interface(self, username: str, customer_id: UUID) -> list[int]: + """ + TODO: how do we prevent a targeted denial-of-service attack? + """ + assert (customer_id in self.customers.keys()) + customer = self.customers[customer_id] + assert (username in customer.users.keys()) + user = customer.users[username] + user.user_interface.shuffle_interface() + return user.user_interface.interface_index def set_nkode( self, username: str, customer_id: UUID, key_selection: list[int], session_id: UUID) -> list[int]: - assert (username not in [user.username for user in self.customers[customer_id].users]) + assert (customer_id in self.customers.keys()) + customer = self.customers[customer_id] + assert (username not in customer.users.keys()) assert (session_id in self.sessions.keys()) - assert (customer_id == self.sessions[session_id].customer_id) - numb_of_keys = self.customers[customer_id].interface.numb_keys - attrs_per_key = self.customers[customer_id].interface.attrs_per_key - assert (all(0 <= key <= numb_of_keys for key in key_selection)) session = self.sessions[session_id] + assert (customer_id == session.customer_id) + numb_of_keys = customer.interface.numb_keys + attrs_per_key = customer.interface.attrs_per_key + assert (all(0 <= key <= numb_of_keys for key in key_selection)) set_interface = UserInterface( interface_index=session.set_interface, numb_sets=attrs_per_key, @@ -82,8 +123,8 @@ class PseudoNKodeAPI(BaseModel): customer = self.customers[customer_id] numb_of_keys = customer.interface.numb_keys attrs_per_key = customer.interface.attrs_per_key - assert(all(0 <= key <= numb_of_keys for key in confirm_key_entry)) - passcode = self._deduce_passcode(session_id, numb_of_keys, attrs_per_key, confirm_key_entry) + assert (all(0 <= key <= numb_of_keys for key in confirm_key_entry)) + passcode = self._deduce_passcode(session_id, attrs_per_key, confirm_key_entry) set_values = customer.interface.set_vals new_user_keys = UserCipherKeys.new_user_encipher_keys(numb_of_keys, attrs_per_key, set_values) enciphered_passcode = new_user_keys.encipher_nkode(passcode, customer.interface) @@ -94,21 +135,22 @@ class PseudoNKodeAPI(BaseModel): user_interface=UserInterface( interface_index=self.sessions[session_id].confirm_interface, numb_sets=attrs_per_key, - numb_keys=numb_of_keys - ) + numb_keys=numb_of_keys, + ), ) - self.customers[customer_id].users.append(new_user) + self.customers[customer_id].add_new_user(new_user) return "success" # del self.sessions[session_id] - def _deduce_passcode(self, session_id: UUID, numb_of_keys: int, attrs_per_key, confirm_key_entry: list[int]) -> list[int]: + def _deduce_passcode(self, session_id: UUID, attrs_per_key, confirm_key_entry: list[int]) -> list[int]: session = self.sessions[session_id] set_key_entry = session.set_key_entry assert (len(set_key_entry) == len(confirm_key_entry)) set_interface = session.set_interface confirm_interface = session.confirm_interface - set_key_vals = [set_interface[key*attrs_per_key:(key+1)*attrs_per_key] for key in set_key_entry] - confirm_key_vals = [confirm_interface[key*attrs_per_key:(key+1)*attrs_per_key] for key in confirm_key_entry] + set_key_vals = [set_interface[key * attrs_per_key:(key + 1) * attrs_per_key] for key in set_key_entry] + confirm_key_vals = [confirm_interface[key * attrs_per_key:(key + 1) * attrs_per_key] for key in + confirm_key_entry] passcode = [] for idx in range(len(set_key_entry)): @@ -119,8 +161,10 @@ class PseudoNKodeAPI(BaseModel): passcode.append(intersection[0]) return passcode - def login(self): - pass + def login(self, customer_id: UUID, username: str, key_selection: list[int]) -> bool: + assert(customer_id in self.customers.keys()) + customer = self.customers[customer_id] + return customer.valid_key_entry(username, key_selection) def renew_keys(self): pass @@ -129,7 +173,7 @@ class PseudoNKodeAPI(BaseModel): new_customer = CustomerDBModel( customer_id=uuid4(), interface=CustomerInterface.new_interface(numb_keys, numb_sets), - users=[], + users={}, ) self.customers[new_customer.customer_id] = new_customer diff --git a/src/user_cipher_keys.py b/src/user_cipher_keys.py index 9b87c15..bc520a5 100644 --- a/src/user_cipher_keys.py +++ b/src/user_cipher_keys.py @@ -6,7 +6,7 @@ from pydantic import BaseModel from src.models import EncipheredNKode from src.nkode_interface import CustomerInterface -from src.utils import generate_random_nonrepeating_list, xor_lists +from src.utils import generate_random_nonrepeating_list, xor_lists, int_array_to_bytes class UserCipherKeys(BaseModel): @@ -15,6 +15,7 @@ class UserCipherKeys(BaseModel): pass_key: list[int] mask_key: list[int] salt: bytes + max_nkode_len: int = 10 @staticmethod def new_user_encipher_keys(numb_of_keys: int, attrs_per_key: int, set_values: list[int]): @@ -24,19 +25,17 @@ class UserCipherKeys(BaseModel): set_key = xor_lists(set_key, set_values) return UserCipherKeys( - alpha_key=generate_random_nonrepeating_list(attrs_per_key*numb_of_keys), + alpha_key=generate_random_nonrepeating_list(attrs_per_key * numb_of_keys), pass_key=generate_random_nonrepeating_list(numb_of_keys), mask_key=generate_random_nonrepeating_list(numb_of_keys), set_key=set_key, salt=bcrypt.gensalt(), ) - @staticmethod - def pad_user_mask(user_mask: list[int], customer_interface: CustomerInterface, max_nkode_len: int) -> list[int]: - assert (len(user_mask) <= max_nkode_len) - set_vals = customer_interface.set_vals + def pad_user_mask(self, user_mask: list[int], set_vals: list[int]) -> list[int]: + assert (len(user_mask) <= self.max_nkode_len) padded_user_mask = user_mask.copy() - for _ in range(max_nkode_len - len(user_mask)): + for _ in range(self.max_nkode_len - len(user_mask)): padded_user_mask.append(choice(set_vals)) return padded_user_mask @@ -47,39 +46,76 @@ class UserCipherKeys(BaseModel): @staticmethod def encode_base64_str(data: list[int]) -> str: - return base64.b64encode(bytes(data)).decode("utf-8") + return base64.b64encode(int_array_to_bytes(data)).decode("utf-8") @staticmethod def decode_base64_str(data: str) -> list[int]: - return list(base64.b64decode(data)) + byte_data = base64.b64decode(data) + int_list = [] + + for i in range(0, len(byte_data), 2): + int_val = int.from_bytes(byte_data[i:i + 2], byteorder='big') + int_list.append(int_val) + return int_list def _hash_passcode(self, passcode: list[int]) -> str: - passcode_digest = base64.b64encode(hashlib.sha256(bytes(passcode)).digest()) + passcode_bytes = int_array_to_bytes(passcode) + passcode_digest = base64.b64encode(hashlib.sha256(passcode_bytes).digest()) hashed_data = bcrypt.hashpw(passcode_digest, self.salt) return hashed_data.decode("utf-8") def encipher_nkode( self, - nkode_attr_index: list[int], + passcode_attr_idx: list[int], customer_interface: CustomerInterface ) -> EncipheredNKode: - max_nkode_len = 10 - passcode_len = len(nkode_attr_index) - user_nkode_attrs = [customer_interface.customer_interface[idx] for idx in nkode_attr_index] - user_nkode_mask = [customer_interface.get_attr_set_val(attr) for attr in user_nkode_attrs] - mask_cipher = self.pad_user_mask(user_nkode_mask, customer_interface, max_nkode_len) + + passcode_attrs = [customer_interface.customer_interface[idx] for idx in passcode_attr_idx] + passcode_sets = [customer_interface.get_attr_set_val(attr) for attr in passcode_attrs] + code = self.encipher_salt_hash_code(passcode_attr_idx, customer_interface) + mask = self.encipher_mask(passcode_sets, customer_interface) + return EncipheredNKode( + code=code, + mask=mask + ) + + def encipher_salt_hash_code( + self, + passcode_attr_idx: list[int], + customer_interface: CustomerInterface, + ) -> str: + passcode_len = len(passcode_attr_idx) + passcode_attrs = [customer_interface.customer_interface[idx] for idx in passcode_attr_idx] passcode_cipher = self.pass_key for idx in range(passcode_len): - attr_idx = nkode_attr_index[idx] + attr_idx = passcode_attr_idx[idx] alpha = self.alpha_key[attr_idx] - attr_val = user_nkode_attrs[idx] + attr_val = passcode_attrs[idx] passcode_cipher[idx] ^= alpha ^ attr_val + return self._hash_passcode(passcode_cipher) + + def encipher_mask( + self, + passcode_sets: list[int], + customer_interface: CustomerInterface + ) -> str: + padded_passcode_sets = self.pad_user_mask(passcode_sets, customer_interface.set_vals) + set_idx = [customer_interface.get_set_index(set_val) for set_val in padded_passcode_sets] + sorted_set_key = [self.set_key[idx] for idx in set_idx] + ciphered_mask = xor_lists(sorted_set_key, padded_passcode_sets) + ciphered_mask = xor_lists(ciphered_mask, self.mask_key) + mask = self.encode_base64_str(ciphered_mask) + return mask + + def decipher_mask(self, mask: str, set_vals: list, passcode_len: int) -> list[int]: + decoded_mask = self.decode_base64_str(mask) + deciphered_mask = xor_lists(decoded_mask, self.mask_key) + set_key_ciphers = xor_lists(set_vals, self.set_key) + passcode_sets = [] + for set_cipher in deciphered_mask[:passcode_len]: + set_idx = set_key_ciphers.index(set_cipher) + passcode_sets.append(set_vals[set_idx]) + return passcode_sets - set_idx = customer_interface.get_set_index(user_nkode_mask[idx]) - mask_cipher[idx] ^= self.set_key[set_idx] ^ self.mask_key[idx] - return EncipheredNKode( - code=self._hash_passcode(passcode_cipher), - mask=self.encode_base64_str(mask_cipher) - ) diff --git a/src/user_interface.py b/src/user_interface.py index 1bc06ac..f341c31 100644 --- a/src/user_interface.py +++ b/src/user_interface.py @@ -1,4 +1,5 @@ from pydantic import BaseModel +from secrets import choice from src.utils import list_to_matrix, secure_fisher_yates_shuffle, matrix_to_list @@ -18,7 +19,7 @@ class UserInterface(BaseModel): def disperse_interface(self): user_interface_matrix = list_to_matrix(self.interface_index, self.numb_sets) shuffled_keys = secure_fisher_yates_shuffle(user_interface_matrix) - dispersed_interface = self._random_attribute_rotation(shuffled_keys) + dispersed_interface = self._random_attribute_rotation(shuffled_keys, list(range(self.numb_sets))) self.interface_index = matrix_to_list(dispersed_interface) @staticmethod @@ -26,13 +27,28 @@ class UserInterface(BaseModel): return [list(row) for row in zip(*interface)] def shuffle_interface(self): - pass + """just like dispersion but only half the sets are rotated""" + numb_of_selected_sets = self.numb_sets // 2 + # randomly shuffle half the sets. if numb_sets is odd, randomly add one 50% of the time + numb_of_selected_sets += choice([0, 1]) if (self.numb_sets & 1) == 1 else 0 + selected_sets = secure_fisher_yates_shuffle(list(range(self.numb_sets)))[:numb_of_selected_sets] + user_interface_matrix = list_to_matrix(self.interface_index, self.numb_sets) + shuffled_keys = secure_fisher_yates_shuffle(user_interface_matrix) + interface_by_sets = [] + for idx, attrs in enumerate(self.matrix_transpose(shuffled_keys)): + if idx in selected_sets: + interface_by_sets.append(secure_fisher_yates_shuffle(attrs)) + else: + interface_by_sets.append(attrs) + self.interface_index = matrix_to_list(self.matrix_transpose(interface_by_sets)) - def _random_attribute_rotation(self, user_interface: list[list[int]]) -> list[list[int]]: + def _random_attribute_rotation(self, user_interface: list[list[int]], selected_sets: list[int]) -> list[list[int]]: attr_rotation = secure_fisher_yates_shuffle(list(range(self.numb_keys)))[:self.numb_sets] transposed_user_interface = self.matrix_transpose(user_interface) assert (len(attr_rotation) == len(transposed_user_interface)) for idx, attr_set in enumerate(transposed_user_interface): + if idx not in selected_sets: + continue rotation = attr_rotation[idx] transposed_user_interface[idx] = attr_set[rotation:] + attr_set[:rotation] return self.matrix_transpose(transposed_user_interface) @@ -46,3 +62,8 @@ class UserInterface(BaseModel): graph[attr].remove(attr) return graph + + def get_key_attr_idxs(self, key_numb: int) -> list[int]: + assert (0 <= key_numb < self.numb_keys) + keypad_attr_idx = list_to_matrix(self.interface_index, self.numb_sets) + return keypad_attr_idx[key_numb] diff --git a/src/utils.py b/src/utils.py index 4c2517a..40321c8 100644 --- a/src/utils.py +++ b/src/utils.py @@ -1,5 +1,6 @@ import secrets + def secure_fisher_yates_shuffle(arr: list) -> list: n = len(arr) for i in range(n - 1, 0, -1): @@ -9,21 +10,8 @@ def secure_fisher_yates_shuffle(arr: list) -> list: def generate_random_nonrepeating_list(list_len: int, min_val: int = 0, max_val: int = 2 ** 16) -> list[int]: - assert(max_val-min_val >= list_len) - return secure_fisher_yates_shuffle(list(range(min_val, max_val))[:list_len]) - - -def generate_random_nonrepeating_matrix(rows: int, cols: int, min_val: int = 0, max_val: int = 2 ** 16) -> list[list[int]]: - values = generate_random_nonrepeating_list(rows*cols, min_val, max_val) - matrix = [] - idx = 0 - for _ in range(cols): - row = [] - for _ in range(rows): - row.append(values[idx]) - idx += 1 - matrix.append(row) - return matrix + assert (max_val - min_val >= list_len) + return secure_fisher_yates_shuffle(list(range(min_val, max_val)))[:list_len] def xor_lists(l1: list[int], l2: list[int]): @@ -31,14 +19,13 @@ def xor_lists(l1: list[int], l2: list[int]): return [l2[i] ^ l1[i] for i in range(len(l1))] -def generate_random_index_interface(height: int, width: int) -> list[int]: - return secure_fisher_yates_shuffle([i for i in range(height * width)]) - - def matrix_to_list(mat: list[list[int]]) -> list[int]: return [val for row in mat for val in row] + def list_to_matrix(lst: list[int], cols: int) -> list[list[int]]: - return [lst[i:i+cols] for i in range(0, len(lst), cols)] + return [lst[i:i + cols] for i in range(0, len(lst), cols)] +def int_array_to_bytes(int_arr: list[int], byte_size: int = 2) -> bytes: + return b"".join([numb.to_bytes(byte_size, byteorder='big') for numb in int_arr]) diff --git a/test/test_encipher.py b/test/test_encipher.py deleted file mode 100644 index d19c352..0000000 --- a/test/test_encipher.py +++ /dev/null @@ -1,5 +0,0 @@ -import pytest - - -def test_encipher_nkode(): - pass diff --git a/test/test_pseudo_nkode_api.py b/test/test_pseudo_nkode_api.py index 5564505..b06bafe 100644 --- a/test/test_pseudo_nkode_api.py +++ b/test/test_pseudo_nkode_api.py @@ -27,3 +27,8 @@ def test_create_new_user(pseudo_nkode_api, numb_keys, attrs_per_key, user_passco session_id ) assert ("success" == response) + + login_interface = pseudo_nkode_api.get_login_index_interface(username, customer.customer_id) + login_key_selection = key_selection(login_interface) + successful_login = pseudo_nkode_api.login(customer.customer_id, username, login_key_selection) + assert (successful_login) diff --git a/test/test_user_cipher_keys.py b/test/test_user_cipher_keys.py new file mode 100644 index 0000000..f484ad6 --- /dev/null +++ b/test/test_user_cipher_keys.py @@ -0,0 +1,35 @@ +import pytest +from src.user_cipher_keys import UserCipherKeys, CustomerInterface +from src.utils import generate_random_nonrepeating_list + + +@pytest.mark.parametrize( + "passcode_len", + [ + 6 + ] +) +def test_encode_decode_base64(passcode_len): + data = generate_random_nonrepeating_list(passcode_len) + encoded = UserCipherKeys.encode_base64_str(data) + decoded = UserCipherKeys.decode_base64_str(encoded) + assert (len(data) == len(decoded)) + assert (all(data[idx] == decoded[idx] for idx in range(passcode_len))) + +@pytest.mark.parametrize( + "numb_of_keys,attrs_per_key", + [ + (10, 7,) + ]) +def test_decode_mask(numb_of_keys, attrs_per_key): + customer = CustomerInterface.new_interface(numb_of_keys, attrs_per_key) + passcode_entry = generate_random_nonrepeating_list(numb_of_keys*attrs_per_key, max_val=70)[:4] + passcode_values = [customer.customer_interface[idx] for idx in passcode_entry] + set_vals = customer.set_vals + user_keys = UserCipherKeys.new_user_encipher_keys(numb_of_keys, attrs_per_key, set_vals) + passcode = user_keys.encipher_nkode(passcode_entry, customer) + + orig_passcode_set_vals = [customer.get_attr_set_val(attr) for attr in passcode_values] + passcode_set_vals = user_keys.decipher_mask(passcode.mask, set_vals, len(passcode_entry)) + assert(len(passcode_set_vals) == len(orig_passcode_set_vals)) + assert(all(orig_passcode_set_vals[idx] == passcode_set_vals[idx] for idx in range(len(passcode_set_vals)))) diff --git a/test/test_user_interface.py b/test/test_user_interface.py index da73637..c677385 100644 --- a/test/test_user_interface.py +++ b/test/test_user_interface.py @@ -1,11 +1,12 @@ import pytest from src.user_interface import UserInterface -@pytest.mark.parametrize("user_interface", [ - ( - UserInterface.new_interface(7, 10) - ) -]) + +@pytest.fixture() +def user_interface(): + return UserInterface.new_interface(7, 10) + + def test_dispersion(user_interface): pre_dispersion_graph = user_interface.attribute_adjacency_graph() user_interface.disperse_interface() @@ -14,3 +15,22 @@ def test_dispersion(user_interface): for _ in range(10000): for attr, adj_graph in pre_dispersion_graph.items(): assert (adj_graph.isdisjoint(post_dispersion_graph[attr])) + + +def test_shuffle_attrs(user_interface): + """there's no easy way to test this. At some point we'll have to run this code thousands of time to see if we get + expected statistical outcomes like: + - every attribute gets to every key with a uniform distribution + - every attribute is adjacent to every other attribute with uniform distribution + - the order in which the attributes move from key to key is random (i.e. the distance traveled is uniform) + """ + pre_shuffle_interface = user_interface.interface_index + user_interface.shuffle_interface() + post_shuffle_interface = user_interface.interface_index + for i in range(1000): + assert (not all( + post_shuffle_interface[idx] == pre_shuffle_interface[idx] for idx in range(len(post_shuffle_interface)) + )) + assert (not all( + post_shuffle_interface[idx] != pre_shuffle_interface[idx] for idx in range(len(post_shuffle_interface)) + ))