diff --git a/src/nkode_interface.py b/src/nkode_interface.py index d5de170..59eb105 100644 --- a/src/nkode_interface.py +++ b/src/nkode_interface.py @@ -1,25 +1,41 @@ from pydantic import BaseModel -from src.utils import generate_random_nonrepeating_list, generate_random_nonrepeating_matrix +from src.utils import ( + generate_random_nonrepeating_list, list_to_matrix, + secure_fisher_yates_shuffle +) -class NKodeInterface(BaseModel): - interface: list[int] +class CustomerInterface(BaseModel): + customer_interface: list[int] set_vals: list[int] numb_keys: int - numb_of_sets: int + attrs_per_key: int @staticmethod - def new_interface(numb_keys: int, numb_of_sets: int): + def new_interface(numb_keys: int, attrs_per_key: int): assert (numb_keys <= 256) - assert (numb_of_sets <= 256) + assert (attrs_per_key <= 256) - return NKodeInterface( - interface=generate_random_nonrepeating_list(numb_of_sets*numb_keys), - sel_vals=generate_random_nonrepeating_list(numb_of_sets), + return CustomerInterface( + customer_interface=generate_random_nonrepeating_list(attrs_per_key*numb_keys), + set_vals=generate_random_nonrepeating_list(attrs_per_key), numb_keys=numb_keys, - numb_of_sets=numb_of_sets, + attrs_per_key=attrs_per_key, ) + def get_interface_by_set(self) -> dict[int, list[int]]: + interface_by_set = {self.set_vals[set_val]: [] for set_val in self.set_vals} + for idx, attr_set in enumerate(self.interface): + set_idx = idx % self.numb_of_sets + interface_by_set[self.set_vals[set_idx]].append(attr_set) + return interface_by_set + + def get_interface_by_key(self) -> list[list[int]]: + return list_to_matrix(self.interface, self.numb_keys) + + def get_random_index_interface(self) -> list[int]: + return secure_fisher_yates_shuffle(list(range(len(self.interface)))) + def get_attr_set_val(self, attr: int) -> int: assert(attr in self.interface) attr_idx = self.interface.index(attr) @@ -28,4 +44,5 @@ class NKodeInterface(BaseModel): def get_set_index(self, set_val: int) -> int: assert(set_val in self.get_set_values()) - return self._set_index_lookup[set_val] + return self.set_vals.index(set_val) + diff --git a/src/pseudo_nkode_api.py b/src/pseudo_nkode_api.py index 5af5421..e058f17 100644 --- a/src/pseudo_nkode_api.py +++ b/src/pseudo_nkode_api.py @@ -1,68 +1,51 @@ from uuid import UUID, uuid4 from pydantic import BaseModel + from src.models import EncipheredNKode from src.user_cipher_keys import UserCipherKeys -from src.nkode_interface import NKodeInterface -from src.utils import generate_random_index_interface, disperse_interface +from src.user_interface import UserInterface +from src.nkode_interface import CustomerInterface class UserDBModel(BaseModel): username: str - nkode: EncipheredNKode + enciphered_passcode: EncipheredNKode user_keys: UserCipherKeys - user_interface: list[list[int]] + user_interface: UserInterface class CustomerDBModel(BaseModel): customer_id: UUID - interface: NKodeInterface + interface: CustomerInterface users: list[UserDBModel] class SessionCacheModel(BaseModel): session_id: UUID - set_interface: list[list[int]] | None - confirm_interface: list[list[int]] | None - set_key_entry: list[int] | None - confirm_key_entry: list[int] | None + set_interface: list[int] | None = None + confirm_interface: list[int] | None = None + set_key_entry: list[int] | None = None + confirm_key_entry: list[int] | None = None + username: str | None = None customer_id: UUID - username: str # timeout: datetime - def create_user(self, customer_interface: NKodeInterface) -> UserDBModel: - assert ( - self.set_interface and - self.confirm_interface and - self.set_key_entry and - self.confirm_key_entry - ) - nkode_attr_indices = self._determine_user_nkode_attr_set_index() - customer_interface.interface - - def _determine_user_nkode_attr_set_index(self) -> list[int]: - set_entry = self.set_key_entry - confirm_entry = self.confirm_key_entry - assert (len(set_entry) == len(confirm_entry)) - set_key_vals = [self.set_interface[key] for key in set_entry] - confirm_key_vals = [self.confirm_interface[key] for key in confirm_entry] - overlap = list(set(set_entry) & set(confirm_entry)) - return overlap - class PseudoNKodeAPI(BaseModel): - numb_of_keys: int = 10 - attrs_per_key: int = 7 - customers: dict[UUID, CustomerDBModel] - sessions: dict[UUID, SessionCacheModel] + customers: dict[UUID, CustomerDBModel] = {} + sessions: dict[UUID, SessionCacheModel] = {} - def generate_index_interface(self, customer_id: UUID) -> tuple[UUID, list[list[int]]]: + def generate_index_interface(self, customer_id: UUID) -> tuple[UUID, list[int]]: assert (customer_id in self.customers.keys()) + customer = self.customers[customer_id] + set_interface = UserInterface.new_interface(customer.interface.attrs_per_key, customer.interface.numb_keys) new_session = SessionCacheModel( session_id=uuid4(), - set_interface=generate_random_index_interface(self.numb_of_keys, self.attrs_per_key), - customer_id=customer_id + set_interface=set_interface.interface_index, + customer_id=customer_id, ) + self.sessions[new_session.session_id] = new_session return new_session.session_id, new_session.set_interface def get_login_index_interface(self): @@ -70,36 +53,68 @@ class PseudoNKodeAPI(BaseModel): def set_nkode( self, username: str, customer_id: UUID, - key_selection: list[int], session_id: UUID) -> list[list[int]]: + key_selection: list[int], session_id: UUID) -> list[int]: assert (username not in [user.username for user in self.customers[customer_id].users]) assert (session_id in self.sessions.keys()) assert (customer_id == self.sessions[session_id].customer_id) - assert (all(0 <= key <= self.numb_of_keys for key in key_selection)) - self.sessions[session_id].username = username - self.sessions[session_id].set_key_entry = key_selection - self.sessions[session_id].confirm_interface = disperse_interface(self.sessions[session_id].set_interface) + 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] + set_interface = UserInterface( + interface_index=session.set_interface, + numb_sets=attrs_per_key, + numb_keys=numb_of_keys, + ) + set_interface.disperse_interface() + session.username = username + session.set_key_entry = key_selection + session.confirm_interface = set_interface.interface_index + self.sessions[session_id] = session return self.sessions[session_id].confirm_interface - def confirm_nkode(self, username: str, customer_id: UUID, key_selection: list[int], session_id: UUID): + def confirm_nkode(self, username: str, customer_id: UUID, confirm_key_entry: list[int], session_id: UUID): assert ( session_id in self.sessions.keys() and customer_id == self.sessions[session_id].customer_id and - username == self.sessions[session_id].username and - all(0 <= key <= self.numb_of_keys for key in key_selection) + username == self.sessions[session_id].username ) 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) attrs_per_key = customer.interface.numb_of_sets 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) new_user = UserDBModel( username=username, - nkode=EncipheredNKode(), + enciphered_passcode=enciphered_passcode, user_keys=new_user_keys, user_interface=self.sessions[session_id].confirm_interface ) self.customers[customer_id].users.append(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]: + 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] + + passcode = [] + for idx in range(len(set_key_entry)): + set_key = set(set_key_vals[idx]) + confirm_key = set(confirm_key_vals[idx]) + intersection = list(set_key.intersection(confirm_key)) + assert (len(intersection) == 1) + passcode.append(intersection[0]) + return passcode def login(self): pass @@ -110,7 +125,7 @@ class PseudoNKodeAPI(BaseModel): def create_new_customer(self, numb_keys: int, numb_sets: int) -> CustomerDBModel: new_customer = CustomerDBModel( customer_id=uuid4(), - interface=NKodeInterface.new_interface(numb_keys, numb_sets), + interface=CustomerInterface.new_interface(numb_keys, numb_sets), users=[], ) self.customers[new_customer.customer_id] = new_customer diff --git a/src/user_cipher_keys.py b/src/user_cipher_keys.py index 8ad830f..ac45664 100644 --- a/src/user_cipher_keys.py +++ b/src/user_cipher_keys.py @@ -1,17 +1,16 @@ import base64 import hashlib - import bcrypt from secrets import choice from pydantic import BaseModel -from src.models import NKodeAttribute, EncipheredNKode -from src.nkode_interface import NKodeInterface -from src.utils import generate_random_nonrepeating_list, xor_lists, generate_random_nonrepeating_matrix +from src.models import EncipheredNKode +from src.nkode_interface import CustomerInterface +from src.utils import generate_random_nonrepeating_list, xor_lists class UserCipherKeys(BaseModel): - alpha_key: list[list[int]] + alpha_key: list[int] set_key: list[int] pass_key: list[int] mask_key: list[int] @@ -25,7 +24,7 @@ class UserCipherKeys(BaseModel): set_key = xor_lists(set_key, set_values) return UserCipherKeys( - alpha_key=generate_random_nonrepeating_matrix(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, @@ -33,9 +32,9 @@ class UserCipherKeys(BaseModel): ) @staticmethod - def pad_user_mask(user_mask: list[int], customer_interface: NKodeInterface, max_nkode_len: int) -> list[int]: + 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_ + set_vals = customer_interface.set_vals padded_user_mask = user_mask.copy() for _ in range(max_nkode_len - len(user_mask)): padded_user_mask.append(choice(set_vals)) @@ -61,20 +60,25 @@ class UserCipherKeys(BaseModel): def encipher_nkode( self, - nkode_attr_index: list[NKodeAttribute], - customer_interface: NKodeInterface + nkode_attr_index: list[int], + customer_interface: CustomerInterface ) -> EncipheredNKode: max_nkode_len = 10 - user_nkode_mask = [attr.set_val for attr in user_nkode_attributes] - user_nkode_attrs = [attr.attr_val for attr in user_nkode_attributes] + passcode_len = len(nkode_attr_index) + user_nkode_attrs = [customer_interface.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_cipher = self.pad_user_code(user_nkode_attrs, max_nkode_len) + passcode_cipher = self.pass_key + + for idx in range(passcode_len): + attr_idx = nkode_attr_index[idx] + alpha = self.alpha_key[attr_idx] + attr_val = user_nkode_attrs[idx] + passcode_cipher[idx] ^= alpha ^ attr_val for idx in range(max_nkode_len): - set_idx = customer_interface.get_set_index(user_nkode_attributes[idx].set_val) - attr_idx = customer_interface.get_set_index(user_nkode_attributes[idx].attr_val) - passcode_cipher ^= self.alpha_key[set_idx][attr_idx] ^ self.pass_key[idx] - mask_cipher ^= self.set_key[set_idx] ^ self.mask_key[idx] + 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), diff --git a/src/user_interface.py b/src/user_interface.py new file mode 100644 index 0000000..1bc06ac --- /dev/null +++ b/src/user_interface.py @@ -0,0 +1,48 @@ +from pydantic import BaseModel +from src.utils import list_to_matrix, secure_fisher_yates_shuffle, matrix_to_list + + +class UserInterface(BaseModel): + interface_index: list[int] + numb_sets: int + numb_keys: int + + @classmethod + def new_interface(cls, numb_sets: int, numb_keys: int): + return UserInterface( + interface_index=secure_fisher_yates_shuffle(list(range(numb_sets*numb_keys))), + numb_sets=numb_sets, + numb_keys=numb_keys, + ) + + 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) + self.interface_index = matrix_to_list(dispersed_interface) + + @staticmethod + def matrix_transpose(interface: list[list[int]]) -> list[list[int]]: + return [list(row) for row in zip(*interface)] + + def shuffle_interface(self): + pass + + def _random_attribute_rotation(self, user_interface: list[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): + rotation = attr_rotation[idx] + transposed_user_interface[idx] = attr_set[rotation:] + attr_set[:rotation] + return self.matrix_transpose(transposed_user_interface) + + def attribute_adjacency_graph(self) -> dict[int, set[int]]: + user_interface_keypad = list_to_matrix(self.interface_index, self.numb_sets) + graph = {} + for key in user_interface_keypad: + for attr in key: + graph[attr] = set(key) + graph[attr].remove(attr) + + return graph diff --git a/src/utils.py b/src/utils.py index 038e914..4c2517a 100644 --- a/src/utils.py +++ b/src/utils.py @@ -1,6 +1,5 @@ import secrets - def secure_fisher_yates_shuffle(arr: list) -> list: n = len(arr) for i in range(n - 1, 0, -1): @@ -32,52 +31,14 @@ 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[list[int]]: - interface = [] - idx = 0 - for i in range(height): - key = [] - for j in range(width): - key.append(idx) - idx += 1 - key = secure_fisher_yates_shuffle(key) - interface.append(key) - return interface +def generate_random_index_interface(height: int, width: int) -> list[int]: + return secure_fisher_yates_shuffle([i for i in range(height * width)]) -def generate_serialized_index_interface(height: int, width: int) -> list[list[int]]: - interface = [] - idx = 0 - for i in range(height): - key = [] - for j in range(width): - key.append(idx) - idx += 1 - interface.append(key) - return interface +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)] -def interface_transpose(interface: list[list[int]]) -> list[list[int]]: - return [list(row) for row in zip(*interface)] - - -def shuffle_interface(self): - pass - - -def disperse_interface(user_interface: list[list[int]]): - shuffled_keys = secure_fisher_yates_shuffle(user_interface) - dispersed_interface = random_attribute_rotation(shuffled_keys) - return dispersed_interface - - -def random_attribute_rotation(user_interface: list[list[int]]) -> list[list[int]]: - numb_attr = len(user_interface[0]) - numb_keys = len(user_interface) - attr_rotation = secure_fisher_yates_shuffle(list(range(numb_keys)))[:numb_attr] - transposed_user_interface = interface_transpose(user_interface) - assert (len(attr_rotation) == len(transposed_user_interface)) - for idx, attr_set in enumerate(transposed_user_interface): - rotation = attr_rotation[idx] - transposed_user_interface[idx] = attr_set[rotation:] + attr_set[:rotation] - return interface_transpose(transposed_user_interface) diff --git a/test/test_nkode_interface.py b/test/test_nkode_interface.py new file mode 100644 index 0000000..e69de29 diff --git a/test/test_pseudo_nkode_api.py b/test/test_pseudo_nkode_api.py index f1ee443..2d5b9a9 100644 --- a/test/test_pseudo_nkode_api.py +++ b/test/test_pseudo_nkode_api.py @@ -1,43 +1,29 @@ import pytest -from src.pseudo_nkode_api import UserDBModel -from src.utils import generate_random_index_interface, disperse_interface -from src.models import EncipheredNKode -from src.user_cipher_keys import UserCipherKeys +from src.pseudo_nkode_api import PseudoNKodeAPI, CustomerDBModel -def attribute_adjacency_graph(interface: list[list[int]]) -> dict[int, set[int]]: - graph = {} - for attr_set in interface: - for attr in attr_set: - graph[attr] = set(attr_set) - graph[attr].remove(attr) - - return graph +@pytest.fixture() +def pseudo_nkode_api() -> PseudoNKodeAPI: + return PseudoNKodeAPI() -@pytest.mark.parametrize("user", [ - ( - UserDBModel( - username="test", - nkode=EncipheredNKode( - code="", - mask="" - ), - user_keys=UserCipherKeys.new_user_encipher_keys( - numb_of_keys=10, - attrs_per_key=7, - set_values=list(range(7)) - ), - user_interface=generate_random_index_interface(10, 7), - ) - ) +@pytest.mark.parametrize("numb_keys,attrs_per_key,user_passcode", [ + (10, 7, [3, 10, 27, 68]), ]) -def test_dispersion(user): - pre_dispersion_interface = user.user_interface - post_dispersion_interface = disperse_interface(user.user_interface) +def test_create_new_user(pseudo_nkode_api, numb_keys, attrs_per_key, user_passcode): + username = "test_username" + customer = pseudo_nkode_api.create_new_customer(numb_keys, attrs_per_key) + session_id, set_interface = pseudo_nkode_api.generate_index_interface(customer.customer_id) - pre_dispresion_graph = attribute_adjacency_graph(pre_dispersion_interface) - post_dispersion_graph = attribute_adjacency_graph(post_dispersion_interface) - for _ in range(10000): - for attr, adj_graph in pre_dispresion_graph.items(): - assert (adj_graph.isdisjoint(post_dispersion_graph[attr])) + key_selection = lambda interface: [interface.index(attr) % attrs_per_key for attr in user_passcode] + set_key_selection = key_selection(set_interface) + + confirm_interface = pseudo_nkode_api.set_nkode(username, customer.customer_id, set_key_selection, session_id) + confirm_key_selection = key_selection(confirm_interface) + response = pseudo_nkode_api.confirm_nkode( + username, + customer.customer_id, + confirm_key_selection, + session_id + ) + assert ("success" == response) diff --git a/test/test_user_interface.py b/test/test_user_interface.py new file mode 100644 index 0000000..da73637 --- /dev/null +++ b/test/test_user_interface.py @@ -0,0 +1,16 @@ +import pytest +from src.user_interface import UserInterface + +@pytest.mark.parametrize("user_interface", [ + ( + UserInterface.new_interface(7, 10) + ) +]) +def test_dispersion(user_interface): + pre_dispersion_graph = user_interface.attribute_adjacency_graph() + user_interface.disperse_interface() + post_dispersion_graph = user_interface.attribute_adjacency_graph() + + for _ in range(10000): + for attr, adj_graph in pre_dispersion_graph.items(): + assert (adj_graph.isdisjoint(post_dispersion_graph[attr]))