implement pseduo nkode api
This commit is contained in:
89
nkode_demo.ipynb
Normal file
89
nkode_demo.ipynb
Normal file
@@ -0,0 +1,89 @@
|
|||||||
|
{
|
||||||
|
"cells": [
|
||||||
|
{
|
||||||
|
"cell_type": "code",
|
||||||
|
"execution_count": 12,
|
||||||
|
"outputs": [],
|
||||||
|
"source": [
|
||||||
|
"import string"
|
||||||
|
],
|
||||||
|
"metadata": {
|
||||||
|
"collapsed": false,
|
||||||
|
"ExecuteTime": {
|
||||||
|
"end_time": "2024-07-10T23:28:52.928913Z",
|
||||||
|
"start_time": "2024-07-10T23:28:52.924646Z"
|
||||||
|
}
|
||||||
|
}
|
||||||
|
},
|
||||||
|
{
|
||||||
|
"cell_type": "code",
|
||||||
|
"execution_count": 26,
|
||||||
|
"outputs": [],
|
||||||
|
"source": [
|
||||||
|
"attributes = string.ascii_letters + string.digits + string.punctuation\n",
|
||||||
|
"attributes = attributes.replace(\"'\", \"\")\n",
|
||||||
|
"attributes = attributes.replace('\"', \"\")\n",
|
||||||
|
"attributes = attributes[:70]"
|
||||||
|
],
|
||||||
|
"metadata": {
|
||||||
|
"collapsed": false,
|
||||||
|
"ExecuteTime": {
|
||||||
|
"end_time": "2024-07-11T11:58:52.615905Z",
|
||||||
|
"start_time": "2024-07-11T11:58:52.614545Z"
|
||||||
|
}
|
||||||
|
}
|
||||||
|
},
|
||||||
|
{
|
||||||
|
"cell_type": "code",
|
||||||
|
"execution_count": 28,
|
||||||
|
"outputs": [
|
||||||
|
{
|
||||||
|
"data": {
|
||||||
|
"text/plain": "70"
|
||||||
|
},
|
||||||
|
"execution_count": 28,
|
||||||
|
"metadata": {},
|
||||||
|
"output_type": "execute_result"
|
||||||
|
}
|
||||||
|
],
|
||||||
|
"source": [],
|
||||||
|
"metadata": {
|
||||||
|
"collapsed": false,
|
||||||
|
"ExecuteTime": {
|
||||||
|
"end_time": "2024-07-11T12:00:03.004548Z",
|
||||||
|
"start_time": "2024-07-11T12:00:03.000564Z"
|
||||||
|
}
|
||||||
|
}
|
||||||
|
},
|
||||||
|
{
|
||||||
|
"cell_type": "code",
|
||||||
|
"execution_count": null,
|
||||||
|
"outputs": [],
|
||||||
|
"source": [],
|
||||||
|
"metadata": {
|
||||||
|
"collapsed": false
|
||||||
|
}
|
||||||
|
}
|
||||||
|
],
|
||||||
|
"metadata": {
|
||||||
|
"kernelspec": {
|
||||||
|
"display_name": "Python 3",
|
||||||
|
"language": "python",
|
||||||
|
"name": "python3"
|
||||||
|
},
|
||||||
|
"language_info": {
|
||||||
|
"codemirror_mode": {
|
||||||
|
"name": "ipython",
|
||||||
|
"version": 2
|
||||||
|
},
|
||||||
|
"file_extension": ".py",
|
||||||
|
"mimetype": "text/x-python",
|
||||||
|
"name": "python",
|
||||||
|
"nbconvert_exporter": "python",
|
||||||
|
"pygments_lexer": "ipython2",
|
||||||
|
"version": "2.7.6"
|
||||||
|
}
|
||||||
|
},
|
||||||
|
"nbformat": 4,
|
||||||
|
"nbformat_minor": 0
|
||||||
|
}
|
||||||
@@ -1 +1,2 @@
|
|||||||
pydantic
|
pydantic
|
||||||
|
bcrypt
|
||||||
|
|||||||
118
src/pseudo_nkode_api.py
Normal file
118
src/pseudo_nkode_api.py
Normal file
@@ -0,0 +1,118 @@
|
|||||||
|
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
|
||||||
|
|
||||||
|
|
||||||
|
class UserDBModel(BaseModel):
|
||||||
|
username: str
|
||||||
|
nkode: EncipheredNKode
|
||||||
|
user_keys: UserCipherKeys
|
||||||
|
user_interface: list[list[int]]
|
||||||
|
|
||||||
|
|
||||||
|
class CustomerDBModel(BaseModel):
|
||||||
|
customer_id: UUID
|
||||||
|
interface: NKodeInterface
|
||||||
|
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
|
||||||
|
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]
|
||||||
|
|
||||||
|
def generate_index_interface(self, customer_id: UUID) -> tuple[UUID, list[list[int]]]:
|
||||||
|
assert (customer_id in self.customers.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
|
||||||
|
)
|
||||||
|
return new_session.session_id, new_session.set_interface
|
||||||
|
|
||||||
|
def get_login_index_interface(self):
|
||||||
|
pass
|
||||||
|
|
||||||
|
def set_nkode(
|
||||||
|
self, username: str, customer_id: UUID,
|
||||||
|
key_selection: list[int], session_id: UUID) -> list[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)
|
||||||
|
return self.sessions[session_id].confirm_interface
|
||||||
|
|
||||||
|
def confirm_nkode(self, username: str, customer_id: UUID, key_selection: 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)
|
||||||
|
)
|
||||||
|
customer = self.customers[customer_id]
|
||||||
|
numb_of_keys = customer.interface.numb_keys
|
||||||
|
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)
|
||||||
|
|
||||||
|
new_user = UserDBModel(
|
||||||
|
username=username,
|
||||||
|
nkode=EncipheredNKode(),
|
||||||
|
user_keys=new_user_keys,
|
||||||
|
user_interface=self.sessions[session_id].confirm_interface
|
||||||
|
)
|
||||||
|
self.customers[customer_id].users.append(new_user)
|
||||||
|
|
||||||
|
def login(self):
|
||||||
|
pass
|
||||||
|
|
||||||
|
def renew_keys(self):
|
||||||
|
pass
|
||||||
|
|
||||||
|
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),
|
||||||
|
users=[],
|
||||||
|
)
|
||||||
|
self.customers[new_customer.customer_id] = new_customer
|
||||||
|
|
||||||
|
return new_customer
|
||||||
43
test/test_pseudo_nkode_api.py
Normal file
43
test/test_pseudo_nkode_api.py
Normal file
@@ -0,0 +1,43 @@
|
|||||||
|
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
|
||||||
|
|
||||||
|
|
||||||
|
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.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),
|
||||||
|
)
|
||||||
|
)
|
||||||
|
])
|
||||||
|
def test_dispersion(user):
|
||||||
|
pre_dispersion_interface = user.user_interface
|
||||||
|
post_dispersion_interface = disperse_interface(user.user_interface)
|
||||||
|
|
||||||
|
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]))
|
||||||
Reference in New Issue
Block a user