refactor user_signup_session.py
This commit is contained in:
52
nkode_api.py
52
nkode_api.py
@@ -3,7 +3,7 @@ from pydantic import BaseModel
|
|||||||
|
|
||||||
from src.customer import Customer
|
from src.customer import Customer
|
||||||
from src.models import NKodePolicy
|
from src.models import NKodePolicy
|
||||||
from src.session_cache import SessionCache
|
from src.user_signup_session import UserSignupSession
|
||||||
from src.user_cipher_keys import UserCipherKeys
|
from src.user_cipher_keys import UserCipherKeys
|
||||||
from src.user import User
|
from src.user import User
|
||||||
from src.user_interface import UserInterface
|
from src.user_interface import UserInterface
|
||||||
@@ -12,13 +12,13 @@ from src.customer_interface import CustomerInterface
|
|||||||
|
|
||||||
class NKodeAPI(BaseModel):
|
class NKodeAPI(BaseModel):
|
||||||
customers: dict[UUID, Customer] = {}
|
customers: dict[UUID, Customer] = {}
|
||||||
sessions: dict[UUID, SessionCache] = {}
|
sessions: dict[UUID, UserSignupSession] = {}
|
||||||
|
|
||||||
def generate_index_interface(self, customer_id: UUID) -> tuple[UUID, list[int]]:
|
def generate_index_interface(self, customer_id: UUID) -> tuple[UUID, list[int]]:
|
||||||
assert (customer_id in self.customers.keys())
|
assert (customer_id in self.customers.keys())
|
||||||
customer = self.customers[customer_id]
|
customer = self.customers[customer_id]
|
||||||
set_interface = UserInterface.new(customer.interface.attrs_per_key, customer.interface.numb_of_keys)
|
set_interface = UserInterface.new(customer.interface.attrs_per_key, customer.interface.numb_of_keys)
|
||||||
new_session = SessionCache(
|
new_session = UserSignupSession(
|
||||||
session_id=uuid4(),
|
session_id=uuid4(),
|
||||||
set_interface=set_interface.attr_indices,
|
set_interface=set_interface.attr_indices,
|
||||||
customer_id=customer_id,
|
customer_id=customer_id,
|
||||||
@@ -38,27 +38,16 @@ class NKodeAPI(BaseModel):
|
|||||||
return user.user_interface.attr_indices
|
return user.user_interface.attr_indices
|
||||||
|
|
||||||
def set_nkode(
|
def set_nkode(
|
||||||
self, username: str, customer_id: UUID,
|
self, username: str,
|
||||||
key_selection: list[int], session_id: UUID) -> list[int]:
|
customer_id: UUID,
|
||||||
|
key_selection: list[int],
|
||||||
|
session_id: UUID
|
||||||
|
) -> list[int]:
|
||||||
assert (customer_id in self.customers.keys())
|
assert (customer_id in self.customers.keys())
|
||||||
customer = self.customers[customer_id]
|
customer = self.customers[customer_id]
|
||||||
assert (username not in customer.users.keys())
|
assert (username not in customer.users.keys())
|
||||||
assert (session_id in self.sessions.keys())
|
assert (session_id in self.sessions.keys())
|
||||||
session = self.sessions[session_id]
|
self.sessions[session_id].set_user_nkode(username, customer, key_selection)
|
||||||
assert (customer_id == session.customer_id)
|
|
||||||
numb_of_keys = customer.interface.numb_of_keys
|
|
||||||
attrs_per_key = customer.interface.attrs_per_key
|
|
||||||
assert (all(0 <= key <= numb_of_keys for key in key_selection))
|
|
||||||
set_interface = UserInterface(
|
|
||||||
attr_indices=session.set_interface,
|
|
||||||
attrs_per_key=attrs_per_key,
|
|
||||||
numb_of_keys=numb_of_keys,
|
|
||||||
)
|
|
||||||
set_interface.disperse_interface()
|
|
||||||
session.username = username
|
|
||||||
session.set_key_entry = key_selection
|
|
||||||
session.confirm_interface = set_interface.attr_indices
|
|
||||||
self.sessions[session_id] = session
|
|
||||||
return self.sessions[session_id].confirm_interface
|
return self.sessions[session_id].confirm_interface
|
||||||
|
|
||||||
def confirm_nkode(self, username: str, customer_id: UUID, confirm_key_entry: list[int], session_id: UUID) -> bool:
|
def confirm_nkode(self, username: str, customer_id: UUID, confirm_key_entry: list[int], session_id: UUID) -> bool:
|
||||||
@@ -67,30 +56,11 @@ class NKodeAPI(BaseModel):
|
|||||||
customer_id == self.sessions[session_id].customer_id and
|
customer_id == self.sessions[session_id].customer_id and
|
||||||
username == self.sessions[session_id].username
|
username == self.sessions[session_id].username
|
||||||
)
|
)
|
||||||
session = self.sessions[session_id]
|
|
||||||
customer = self.customers[customer_id]
|
customer = self.customers[customer_id]
|
||||||
numb_of_keys = customer.interface.numb_of_keys
|
new_user = self.sessions[session_id].create_user(username, customer, confirm_key_entry)
|
||||||
attrs_per_key = customer.interface.attrs_per_key
|
|
||||||
assert (all(0 <= key <= numb_of_keys for key in confirm_key_entry))
|
|
||||||
passcode = session.deduce_passcode(attrs_per_key, confirm_key_entry)
|
|
||||||
set_values = customer.interface.set_vals
|
|
||||||
if not customer.valid_new_nkode(passcode):
|
|
||||||
return False
|
|
||||||
new_user_keys = UserCipherKeys.new(numb_of_keys, attrs_per_key, set_values, customer.nkode_policy.max_nkode_len)
|
|
||||||
enciphered_passcode = new_user_keys.encipher_nkode(passcode, customer.interface)
|
|
||||||
new_user = User(
|
|
||||||
username=username,
|
|
||||||
enciphered_passcode=enciphered_passcode,
|
|
||||||
user_keys=new_user_keys,
|
|
||||||
user_interface=UserInterface(
|
|
||||||
attr_indices=self.sessions[session_id].confirm_interface,
|
|
||||||
attrs_per_key=attrs_per_key,
|
|
||||||
numb_of_keys=numb_of_keys,
|
|
||||||
),
|
|
||||||
)
|
|
||||||
self.customers[customer_id].add_new_user(new_user)
|
self.customers[customer_id].add_new_user(new_user)
|
||||||
|
del self.sessions[session_id]
|
||||||
return True
|
return True
|
||||||
# del self.sessions[session_id]
|
|
||||||
|
|
||||||
def login(self, customer_id: UUID, username: str, key_selection: list[int]) -> bool:
|
def login(self, customer_id: UUID, username: str, key_selection: list[int]) -> bool:
|
||||||
assert (customer_id in self.customers.keys())
|
assert (customer_id in self.customers.keys())
|
||||||
|
|||||||
@@ -1,5 +1,5 @@
|
|||||||
from pydantic import BaseModel
|
from pydantic import BaseModel
|
||||||
from src.utils import generate_random_nonrepeating_list
|
from src.utils import generate_random_nonrepeating_list, list_to_matrix, matrix_transpose
|
||||||
|
|
||||||
|
|
||||||
class CustomerInterface(BaseModel):
|
class CustomerInterface(BaseModel):
|
||||||
|
|||||||
@@ -1,31 +0,0 @@
|
|||||||
from uuid import UUID
|
|
||||||
|
|
||||||
from pydantic import BaseModel
|
|
||||||
|
|
||||||
|
|
||||||
class SessionCache(BaseModel):
|
|
||||||
session_id: UUID
|
|
||||||
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
|
|
||||||
|
|
||||||
def deduce_passcode(self, attrs_per_key, confirm_key_entry: list[int]) -> list[int]:
|
|
||||||
set_key_entry = self.set_key_entry
|
|
||||||
assert (len(set_key_entry) == len(confirm_key_entry))
|
|
||||||
set_interface = self.set_interface
|
|
||||||
confirm_interface = self.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
|
|
||||||
@@ -1,6 +1,6 @@
|
|||||||
from pydantic import BaseModel
|
from pydantic import BaseModel
|
||||||
from secrets import choice
|
from secrets import choice
|
||||||
from src.utils import list_to_matrix, secure_fisher_yates_shuffle, matrix_to_list
|
from src.utils import list_to_matrix, secure_fisher_yates_shuffle, matrix_to_list, matrix_transpose
|
||||||
|
|
||||||
|
|
||||||
class UserInterface(BaseModel):
|
class UserInterface(BaseModel):
|
||||||
@@ -27,10 +27,6 @@ class UserInterface(BaseModel):
|
|||||||
dispersed_interface = self._random_attribute_rotation(shuffled_keys, list(range(self.attrs_per_key)))
|
dispersed_interface = self._random_attribute_rotation(shuffled_keys, list(range(self.attrs_per_key)))
|
||||||
self.attr_indices = matrix_to_list(dispersed_interface)
|
self.attr_indices = 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):
|
def shuffle_interface(self):
|
||||||
"""just like dispersion but only half the sets are rotated"""
|
"""just like dispersion but only half the sets are rotated"""
|
||||||
numb_of_selected_sets = self.attrs_per_key // 2
|
numb_of_selected_sets = self.attrs_per_key // 2
|
||||||
@@ -40,23 +36,23 @@ class UserInterface(BaseModel):
|
|||||||
user_interface_matrix = list_to_matrix(self.attr_indices, self.attrs_per_key)
|
user_interface_matrix = list_to_matrix(self.attr_indices, self.attrs_per_key)
|
||||||
shuffled_keys = secure_fisher_yates_shuffle(user_interface_matrix)
|
shuffled_keys = secure_fisher_yates_shuffle(user_interface_matrix)
|
||||||
interface_by_sets = []
|
interface_by_sets = []
|
||||||
for idx, attrs in enumerate(self.matrix_transpose(shuffled_keys)):
|
for idx, attrs in enumerate(matrix_transpose(shuffled_keys)):
|
||||||
if idx in selected_sets:
|
if idx in selected_sets:
|
||||||
interface_by_sets.append(secure_fisher_yates_shuffle(attrs))
|
interface_by_sets.append(secure_fisher_yates_shuffle(attrs))
|
||||||
else:
|
else:
|
||||||
interface_by_sets.append(attrs)
|
interface_by_sets.append(attrs)
|
||||||
self.attr_indices = matrix_to_list(self.matrix_transpose(interface_by_sets))
|
self.attr_indices = matrix_to_list(matrix_transpose(interface_by_sets))
|
||||||
|
|
||||||
def _random_attribute_rotation(self, user_interface: list[list[int]], selected_sets: 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_of_keys)))[:self.attrs_per_key]
|
attr_rotation = secure_fisher_yates_shuffle(list(range(self.numb_of_keys)))[:self.attrs_per_key]
|
||||||
transposed_user_interface = self.matrix_transpose(user_interface)
|
transposed_user_interface = matrix_transpose(user_interface)
|
||||||
assert (len(attr_rotation) == len(transposed_user_interface))
|
assert (len(attr_rotation) == len(transposed_user_interface))
|
||||||
for idx, attr_set in enumerate(transposed_user_interface):
|
for idx, attr_set in enumerate(transposed_user_interface):
|
||||||
if idx not in selected_sets:
|
if idx not in selected_sets:
|
||||||
continue
|
continue
|
||||||
rotation = attr_rotation[idx]
|
rotation = attr_rotation[idx]
|
||||||
transposed_user_interface[idx] = attr_set[rotation:] + attr_set[:rotation]
|
transposed_user_interface[idx] = attr_set[rotation:] + attr_set[:rotation]
|
||||||
return self.matrix_transpose(transposed_user_interface)
|
return matrix_transpose(transposed_user_interface)
|
||||||
|
|
||||||
def attribute_adjacency_graph(self) -> dict[int, set[int]]:
|
def attribute_adjacency_graph(self) -> dict[int, set[int]]:
|
||||||
user_interface_keypad = list_to_matrix(self.attr_indices, self.attrs_per_key)
|
user_interface_keypad = list_to_matrix(self.attr_indices, self.attrs_per_key)
|
||||||
|
|||||||
71
src/user_signup_session.py
Normal file
71
src/user_signup_session.py
Normal file
@@ -0,0 +1,71 @@
|
|||||||
|
from uuid import UUID
|
||||||
|
|
||||||
|
from pydantic import BaseModel
|
||||||
|
|
||||||
|
from src.customer import Customer
|
||||||
|
from src.user import User
|
||||||
|
from src.user_cipher_keys import UserCipherKeys
|
||||||
|
from src.user_interface import UserInterface
|
||||||
|
|
||||||
|
|
||||||
|
class UserSignupSession(BaseModel):
|
||||||
|
session_id: UUID
|
||||||
|
customer_id: UUID
|
||||||
|
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
|
||||||
|
|
||||||
|
def _deduce_passcode(self, attrs_per_key, confirm_key_entry: list[int]) -> list[int]:
|
||||||
|
set_key_entry = self.set_key_entry
|
||||||
|
assert (len(set_key_entry) == len(confirm_key_entry))
|
||||||
|
set_interface = self.set_interface
|
||||||
|
confirm_interface = self.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 set_user_nkode(self, username: str, customer: Customer, key_selection: list[int]):
|
||||||
|
assert (customer.customer_id == self.customer_id)
|
||||||
|
numb_of_keys = customer.interface.numb_of_keys
|
||||||
|
attrs_per_key = customer.interface.attrs_per_key
|
||||||
|
assert (all(0 <= key <= numb_of_keys for key in key_selection))
|
||||||
|
set_interface = UserInterface(
|
||||||
|
attr_indices=self.set_interface,
|
||||||
|
attrs_per_key=attrs_per_key,
|
||||||
|
numb_of_keys=numb_of_keys,
|
||||||
|
)
|
||||||
|
set_interface.disperse_interface()
|
||||||
|
self.username = username
|
||||||
|
self.set_key_entry = key_selection
|
||||||
|
self.confirm_interface = set_interface.attr_indices
|
||||||
|
|
||||||
|
def create_user(self, username: str, customer: Customer, confirm_key_entry: list[int]) -> User:
|
||||||
|
numb_of_keys = customer.interface.numb_of_keys
|
||||||
|
attrs_per_key = customer.interface.attrs_per_key
|
||||||
|
assert (all(0 <= key <= numb_of_keys for key in confirm_key_entry))
|
||||||
|
passcode_attrs = self._deduce_passcode(attrs_per_key, confirm_key_entry)
|
||||||
|
set_values = customer.interface.set_vals
|
||||||
|
assert customer.valid_new_nkode(passcode_attrs)
|
||||||
|
new_user_keys = UserCipherKeys.new(numb_of_keys, attrs_per_key, set_values, customer.nkode_policy.max_nkode_len)
|
||||||
|
enciphered_passcode = new_user_keys.encipher_nkode(passcode_attrs, customer.interface)
|
||||||
|
return User(
|
||||||
|
username=username,
|
||||||
|
enciphered_passcode=enciphered_passcode,
|
||||||
|
user_keys=new_user_keys,
|
||||||
|
user_interface=UserInterface(
|
||||||
|
attr_indices=self.confirm_interface,
|
||||||
|
attrs_per_key=attrs_per_key,
|
||||||
|
numb_of_keys=numb_of_keys,
|
||||||
|
),
|
||||||
|
)
|
||||||
@@ -27,5 +27,9 @@ 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 matrix_transpose(interface: list[list[int]]) -> list[list[int]]:
|
||||||
|
return [list(row) for row in zip(*interface)]
|
||||||
|
|
||||||
|
|
||||||
def int_array_to_bytes(int_arr: list[int], byte_size: int = 2) -> bytes:
|
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])
|
return b"".join([numb.to_bytes(byte_size, byteorder='big') for numb in int_arr])
|
||||||
|
|||||||
Reference in New Issue
Block a user