From 0759518dce68b4cc03a14bb4396a7709a9d9897d Mon Sep 17 00:00:00 2001 From: mirai Date: Fri, 7 Nov 2025 16:33:18 +0530 Subject: [PATCH] Remove .history folder --- .../cmd_chat/client/client_20251105191919.py | 162 ------------------ .../cmd_chat/client/client_20251105191921.py | 162 ------------------ .../core/abs/abs_crypto_20251105190705.py | 27 --- .../core/abs/abs_crypto_20251105190714.py | 27 --- .../core/abs/abs_renderer_20251105190654.py | 26 --- .../core/abs/abs_renderer_20251105190714.py | 26 --- .../core/abs/abs_renderer_20251105191905.py | 34 ---- .../core/abs/abs_renderer_20251105191906.py | 34 ---- .../client/core/crypto_20251105190754.py | 44 ----- .../client/core/crypto_20251105190758.py | 44 ----- .../core/default_renderer_20251105190717.py | 61 ------- .../core/default_renderer_20251105190719.py | 61 ------- .../core/default_renderer_20251105191806.py | 61 ------- .../core/default_renderer_20251105191832.py | 61 ------- .../core/default_renderer_20251105191906.py | 61 ------- .../core/rich_renderer_20251105190741.py | 78 --------- .../core/rich_renderer_20251105190758.py | 78 --------- .../core/rich_renderer_20251105191843.py | 78 --------- .../core/rich_renderer_20251105191852.py | 78 --------- .../core/rich_renderer_20251105191906.py | 78 --------- .../cmd_chat/server/server_20251105191755.py | 113 ------------ .../cmd_chat/server/server_20251105191907.py | 113 ------------ .../server/services_20251105191745.py | 36 ---- .../server/services_20251105191746.py | 36 ---- .history/setup_20251105190626.py | 33 ---- .history/setup_20251105190632.py | 33 ---- 26 files changed, 1645 deletions(-) delete mode 100644 .history/cmd_chat/client/client_20251105191919.py delete mode 100644 .history/cmd_chat/client/client_20251105191921.py delete mode 100644 .history/cmd_chat/client/core/abs/abs_crypto_20251105190705.py delete mode 100644 .history/cmd_chat/client/core/abs/abs_crypto_20251105190714.py delete mode 100644 .history/cmd_chat/client/core/abs/abs_renderer_20251105190654.py delete mode 100644 .history/cmd_chat/client/core/abs/abs_renderer_20251105190714.py delete mode 100644 .history/cmd_chat/client/core/abs/abs_renderer_20251105191905.py delete mode 100644 .history/cmd_chat/client/core/abs/abs_renderer_20251105191906.py delete mode 100644 .history/cmd_chat/client/core/crypto_20251105190754.py delete mode 100644 .history/cmd_chat/client/core/crypto_20251105190758.py delete mode 100644 .history/cmd_chat/client/core/default_renderer_20251105190717.py delete mode 100644 .history/cmd_chat/client/core/default_renderer_20251105190719.py delete mode 100644 .history/cmd_chat/client/core/default_renderer_20251105191806.py delete mode 100644 .history/cmd_chat/client/core/default_renderer_20251105191832.py delete mode 100644 .history/cmd_chat/client/core/default_renderer_20251105191906.py delete mode 100644 .history/cmd_chat/client/core/rich_renderer_20251105190741.py delete mode 100644 .history/cmd_chat/client/core/rich_renderer_20251105190758.py delete mode 100644 .history/cmd_chat/client/core/rich_renderer_20251105191843.py delete mode 100644 .history/cmd_chat/client/core/rich_renderer_20251105191852.py delete mode 100644 .history/cmd_chat/client/core/rich_renderer_20251105191906.py delete mode 100644 .history/cmd_chat/server/server_20251105191755.py delete mode 100644 .history/cmd_chat/server/server_20251105191907.py delete mode 100644 .history/cmd_chat/server/services_20251105191745.py delete mode 100644 .history/cmd_chat/server/services_20251105191746.py delete mode 100644 .history/setup_20251105190626.py delete mode 100644 .history/setup_20251105190632.py diff --git a/.history/cmd_chat/client/client_20251105191919.py b/.history/cmd_chat/client/client_20251105191919.py deleted file mode 100644 index b50bdc9..0000000 --- a/.history/cmd_chat/client/client_20251105191919.py +++ /dev/null @@ -1,162 +0,0 @@ -import ast -import time -import threading -from typing import Optional - -from websocket import create_connection, WebSocketConnectionClosedException - -from cmd_chat.client.core.crypto import RSAService -from cmd_chat.client.core.default_renderer import DefaultClientRenderer -from cmd_chat.client.core.rich_renderer import RichClientRenderer - -from cmd_chat.client.config import RENDER_TIME - - -class Client(RSAService, RichClientRenderer): - - def __init__( - self, - server: str, - port: int, - username: str, - password: Optional[str] = None - ): - super().__init__() - self.server = server - self.port = port - self.username = username - self.password = password or "" - self.base_url = f"http://{self.server}:{self.port}" - self.ws_url = f"ws://{self.server}:{self.port}" - self.close_response = str({ - "action": "close", - "username": self.username - }) - self.__stop_threads = False - - def _ws_full(self, path: str) -> str: - if self.password: - return f"{self.ws_url}{path}?password={self.password}" - return f"{self.ws_url}{path}" - - def _connect_ws(self, path: str, retries: int = 5, backoff: float = 0.5): - last_exc: Exception = ConnectionError("Failed to connect") - for attempt in range(retries): - try: - return create_connection(self._ws_full(path)) - except Exception as exc: - last_exc = exc - time.sleep(backoff * (2 ** attempt)) - print(f"Can't connect to {path}: {last_exc}") - raise last_exc - - def send_info(self): - ws = self._connect_ws("/talk") - try: - while not self.__stop_threads: - try: - user_input = input("You're message: ") - if user_input == "q": - self.__stop_threads = True - try: - if ws: - ws.send(self.close_response) - ws.close() - except Exception: - pass - break - message = f'{self.username}: {user_input}' - socket_message = str({ - "text": self._encrypt(message), - "username": self.username - }) - ws.send(socket_message) - except (WebSocketConnectionClosedException, ConnectionResetError, ConnectionAbortedError, OSError): - try: - if ws: - try: - ws.close() - except Exception: - pass - ws = self._connect_ws("/talk") - continue - except Exception: - print("Can't establish channel") - self.__stop_threads = True - break - except KeyboardInterrupt: - self.__stop_threads = True - try: - ws.send(self.close_response) - ws.close() - except Exception: - pass - break - finally: - try: - ws.close() - except Exception: - pass - - def update_info(self): - ws = self._connect_ws("/update") - last_try = None - try: - while not self.__stop_threads: - try: - time.sleep(RENDER_TIME) - raw = ws.recv() - if isinstance(raw, bytes): - raw = raw.decode("utf-8") - response = ast.literal_eval(raw) - if last_try == response: - continue - last_try = response - self.clear_console() - if len(last_try["messages"]) > 0: - self.print_chat(response=last_try) - except (WebSocketConnectionClosedException, ConnectionResetError, ConnectionAbortedError, OSError): - try: - if ws: - try: - ws.close() - except Exception: - pass - ws = self._connect_ws("/update") - continue - except Exception: - print("Connection lost: can't establish update channel") - self.__stop_threads = True - break - except KeyboardInterrupt: - self.__stop_threads = True - try: - ws.send(self.close_response) - ws.close() - except Exception: - pass - break - finally: - try: - ws.close() - except Exception: - pass - - def _validate_keys(self) -> None: - self._request_key( - url=f"{self.base_url}/get_key", - username=self.username, - password=self.password - ) - self._remove_keys() - - def run(self): - self._validate_keys() - threads = [ - threading.Thread(target=self.send_info, daemon=True), - threading.Thread(target=self.update_info, daemon=True) - ] - for th in threads: - th.start() - for th in threads: - th.join() diff --git a/.history/cmd_chat/client/client_20251105191921.py b/.history/cmd_chat/client/client_20251105191921.py deleted file mode 100644 index b50bdc9..0000000 --- a/.history/cmd_chat/client/client_20251105191921.py +++ /dev/null @@ -1,162 +0,0 @@ -import ast -import time -import threading -from typing import Optional - -from websocket import create_connection, WebSocketConnectionClosedException - -from cmd_chat.client.core.crypto import RSAService -from cmd_chat.client.core.default_renderer import DefaultClientRenderer -from cmd_chat.client.core.rich_renderer import RichClientRenderer - -from cmd_chat.client.config import RENDER_TIME - - -class Client(RSAService, RichClientRenderer): - - def __init__( - self, - server: str, - port: int, - username: str, - password: Optional[str] = None - ): - super().__init__() - self.server = server - self.port = port - self.username = username - self.password = password or "" - self.base_url = f"http://{self.server}:{self.port}" - self.ws_url = f"ws://{self.server}:{self.port}" - self.close_response = str({ - "action": "close", - "username": self.username - }) - self.__stop_threads = False - - def _ws_full(self, path: str) -> str: - if self.password: - return f"{self.ws_url}{path}?password={self.password}" - return f"{self.ws_url}{path}" - - def _connect_ws(self, path: str, retries: int = 5, backoff: float = 0.5): - last_exc: Exception = ConnectionError("Failed to connect") - for attempt in range(retries): - try: - return create_connection(self._ws_full(path)) - except Exception as exc: - last_exc = exc - time.sleep(backoff * (2 ** attempt)) - print(f"Can't connect to {path}: {last_exc}") - raise last_exc - - def send_info(self): - ws = self._connect_ws("/talk") - try: - while not self.__stop_threads: - try: - user_input = input("You're message: ") - if user_input == "q": - self.__stop_threads = True - try: - if ws: - ws.send(self.close_response) - ws.close() - except Exception: - pass - break - message = f'{self.username}: {user_input}' - socket_message = str({ - "text": self._encrypt(message), - "username": self.username - }) - ws.send(socket_message) - except (WebSocketConnectionClosedException, ConnectionResetError, ConnectionAbortedError, OSError): - try: - if ws: - try: - ws.close() - except Exception: - pass - ws = self._connect_ws("/talk") - continue - except Exception: - print("Can't establish channel") - self.__stop_threads = True - break - except KeyboardInterrupt: - self.__stop_threads = True - try: - ws.send(self.close_response) - ws.close() - except Exception: - pass - break - finally: - try: - ws.close() - except Exception: - pass - - def update_info(self): - ws = self._connect_ws("/update") - last_try = None - try: - while not self.__stop_threads: - try: - time.sleep(RENDER_TIME) - raw = ws.recv() - if isinstance(raw, bytes): - raw = raw.decode("utf-8") - response = ast.literal_eval(raw) - if last_try == response: - continue - last_try = response - self.clear_console() - if len(last_try["messages"]) > 0: - self.print_chat(response=last_try) - except (WebSocketConnectionClosedException, ConnectionResetError, ConnectionAbortedError, OSError): - try: - if ws: - try: - ws.close() - except Exception: - pass - ws = self._connect_ws("/update") - continue - except Exception: - print("Connection lost: can't establish update channel") - self.__stop_threads = True - break - except KeyboardInterrupt: - self.__stop_threads = True - try: - ws.send(self.close_response) - ws.close() - except Exception: - pass - break - finally: - try: - ws.close() - except Exception: - pass - - def _validate_keys(self) -> None: - self._request_key( - url=f"{self.base_url}/get_key", - username=self.username, - password=self.password - ) - self._remove_keys() - - def run(self): - self._validate_keys() - threads = [ - threading.Thread(target=self.send_info, daemon=True), - threading.Thread(target=self.update_info, daemon=True) - ] - for th in threads: - th.start() - for th in threads: - th.join() diff --git a/.history/cmd_chat/client/core/abs/abs_crypto_20251105190705.py b/.history/cmd_chat/client/core/abs/abs_crypto_20251105190705.py deleted file mode 100644 index e9ab4fb..0000000 --- a/.history/cmd_chat/client/core/abs/abs_crypto_20251105190705.py +++ /dev/null @@ -1,27 +0,0 @@ -from abc import ABC, abstractmethod - -class CryptoService(ABC): - - @abstractmethod - def _encrypt(self, message: str) -> str: - raise NotImplementedError("Need to implement encrypt method") - - @abstractmethod - def _decrypt(self, message: str) -> str: - raise NotImplementedError("Need to implement decrypt method") - - @abstractmethod - def _request_key(self, url: str, username: str, password: str | None = None): - raise NotImplementedError("Need to implement request key method") - - @abstractmethod - def _generate_keys(self): - raise NotImplementedError("Need to implement generate keys method") - - @abstractmethod - def _get_generated_keys(self) -> tuple: - raise NotImplementedError("Need to implement get generated keys method") - - @abstractmethod - def _remove_keys(self): - raise NotImplementedError("Need to implement remove keys method") diff --git a/.history/cmd_chat/client/core/abs/abs_crypto_20251105190714.py b/.history/cmd_chat/client/core/abs/abs_crypto_20251105190714.py deleted file mode 100644 index e9ab4fb..0000000 --- a/.history/cmd_chat/client/core/abs/abs_crypto_20251105190714.py +++ /dev/null @@ -1,27 +0,0 @@ -from abc import ABC, abstractmethod - -class CryptoService(ABC): - - @abstractmethod - def _encrypt(self, message: str) -> str: - raise NotImplementedError("Need to implement encrypt method") - - @abstractmethod - def _decrypt(self, message: str) -> str: - raise NotImplementedError("Need to implement decrypt method") - - @abstractmethod - def _request_key(self, url: str, username: str, password: str | None = None): - raise NotImplementedError("Need to implement request key method") - - @abstractmethod - def _generate_keys(self): - raise NotImplementedError("Need to implement generate keys method") - - @abstractmethod - def _get_generated_keys(self) -> tuple: - raise NotImplementedError("Need to implement get generated keys method") - - @abstractmethod - def _remove_keys(self): - raise NotImplementedError("Need to implement remove keys method") diff --git a/.history/cmd_chat/client/core/abs/abs_renderer_20251105190654.py b/.history/cmd_chat/client/core/abs/abs_renderer_20251105190654.py deleted file mode 100644 index e13f4f0..0000000 --- a/.history/cmd_chat/client/core/abs/abs_renderer_20251105190654.py +++ /dev/null @@ -1,26 +0,0 @@ -from abc import ABC, abstractmethod - - -class ClientRenderer(ABC): - - @abstractmethod - def print_message(self, message: str) -> str: - raise NotImplementedError("Need to implement print_message") - - @abstractmethod - def clear_console(self) -> None: - """Clear the client console (platform-specific).""" - raise NotImplementedError("Need to implement clear_console") - - @abstractmethod - def print_ip(self, ip: str) -> str: - raise NotImplementedError("Need to implement print_ip") - - @abstractmethod - def print_username(self, username: str) -> str: - raise NotImplementedError("Need to implement print_username") - - @abstractmethod - def print_chat(self, response) -> None: - """Render chat payload (response is expected to be a mapping with 'messages' and 'users_in_chat').""" - raise NotImplementedError("Need to implement print_chat") diff --git a/.history/cmd_chat/client/core/abs/abs_renderer_20251105190714.py b/.history/cmd_chat/client/core/abs/abs_renderer_20251105190714.py deleted file mode 100644 index e13f4f0..0000000 --- a/.history/cmd_chat/client/core/abs/abs_renderer_20251105190714.py +++ /dev/null @@ -1,26 +0,0 @@ -from abc import ABC, abstractmethod - - -class ClientRenderer(ABC): - - @abstractmethod - def print_message(self, message: str) -> str: - raise NotImplementedError("Need to implement print_message") - - @abstractmethod - def clear_console(self) -> None: - """Clear the client console (platform-specific).""" - raise NotImplementedError("Need to implement clear_console") - - @abstractmethod - def print_ip(self, ip: str) -> str: - raise NotImplementedError("Need to implement print_ip") - - @abstractmethod - def print_username(self, username: str) -> str: - raise NotImplementedError("Need to implement print_username") - - @abstractmethod - def print_chat(self, response) -> None: - """Render chat payload (response is expected to be a mapping with 'messages' and 'users_in_chat').""" - raise NotImplementedError("Need to implement print_chat") diff --git a/.history/cmd_chat/client/core/abs/abs_renderer_20251105191905.py b/.history/cmd_chat/client/core/abs/abs_renderer_20251105191905.py deleted file mode 100644 index 38ef96e..0000000 --- a/.history/cmd_chat/client/core/abs/abs_renderer_20251105191905.py +++ /dev/null @@ -1,34 +0,0 @@ -from abc import ABC, abstractmethod - - -class ClientRenderer(ABC): - # These attributes are expected to be provided by subclasses - # (typically via multiple inheritance with CryptoService) - username: str - - @abstractmethod - def _decrypt(self, message: str) -> str: - """Decrypt an encrypted message (provided by crypto mixin).""" - raise NotImplementedError("Need to implement _decrypt") - - @abstractmethod - def print_message(self, message: str) -> str: - raise NotImplementedError("Need to implement print_message") - - @abstractmethod - def clear_console(self) -> None: - """Clear the client console (platform-specific).""" - raise NotImplementedError("Need to implement clear_console") - - @abstractmethod - def print_ip(self, ip: str) -> str: - raise NotImplementedError("Need to implement print_ip") - - @abstractmethod - def print_username(self, username: str) -> str: - raise NotImplementedError("Need to implement print_username") - - @abstractmethod - def print_chat(self, response) -> None: - """Render chat payload (response is expected to be a mapping with 'messages' and 'users_in_chat').""" - raise NotImplementedError("Need to implement print_chat") diff --git a/.history/cmd_chat/client/core/abs/abs_renderer_20251105191906.py b/.history/cmd_chat/client/core/abs/abs_renderer_20251105191906.py deleted file mode 100644 index 38ef96e..0000000 --- a/.history/cmd_chat/client/core/abs/abs_renderer_20251105191906.py +++ /dev/null @@ -1,34 +0,0 @@ -from abc import ABC, abstractmethod - - -class ClientRenderer(ABC): - # These attributes are expected to be provided by subclasses - # (typically via multiple inheritance with CryptoService) - username: str - - @abstractmethod - def _decrypt(self, message: str) -> str: - """Decrypt an encrypted message (provided by crypto mixin).""" - raise NotImplementedError("Need to implement _decrypt") - - @abstractmethod - def print_message(self, message: str) -> str: - raise NotImplementedError("Need to implement print_message") - - @abstractmethod - def clear_console(self) -> None: - """Clear the client console (platform-specific).""" - raise NotImplementedError("Need to implement clear_console") - - @abstractmethod - def print_ip(self, ip: str) -> str: - raise NotImplementedError("Need to implement print_ip") - - @abstractmethod - def print_username(self, username: str) -> str: - raise NotImplementedError("Need to implement print_username") - - @abstractmethod - def print_chat(self, response) -> None: - """Render chat payload (response is expected to be a mapping with 'messages' and 'users_in_chat').""" - raise NotImplementedError("Need to implement print_chat") diff --git a/.history/cmd_chat/client/core/crypto_20251105190754.py b/.history/cmd_chat/client/core/crypto_20251105190754.py deleted file mode 100644 index 976e7ab..0000000 --- a/.history/cmd_chat/client/core/crypto_20251105190754.py +++ /dev/null @@ -1,44 +0,0 @@ -import rsa -import requests -from cryptography.fernet import Fernet - -from cmd_chat.client.core.abs.abs_crypto import CryptoService - - -class RSAService(CryptoService): - def __init__(self): - self.public_key = None - self.private_key = None - self.symmetric_key = None - self.fernet = None - self._generate_keys() - - def _encrypt(self, message: str) -> str: - return self.fernet.encrypt(message.encode()).decode("utf-8") - - def _decrypt(self, message: str) -> str: - return self.fernet.decrypt(message.encode()).decode("utf-8") - - def _request_key(self, url: str, username: str, password: str | None = None): - pubkey_bytes = self.public_key.save_pkcs1() - r = requests.post( - url, - files={"pubkey": ("public.pem", pubkey_bytes)}, - data={"username": username, "password": password or ""}, - stream=True, - ) - r.raise_for_status() - # read the full response content (server returns encrypted symmetric key) - message = r.content - self.symmetric_key = rsa.decrypt(message, self.private_key) - self.fernet = Fernet(self.symmetric_key) - - def _generate_keys(self): - self.public_key, self.private_key = rsa.newkeys(512) - - def _get_generated_keys(self): - return self.private_key, self.public_key - - def _remove_keys(self): - self.public_key = None - self.private_key = None \ No newline at end of file diff --git a/.history/cmd_chat/client/core/crypto_20251105190758.py b/.history/cmd_chat/client/core/crypto_20251105190758.py deleted file mode 100644 index 976e7ab..0000000 --- a/.history/cmd_chat/client/core/crypto_20251105190758.py +++ /dev/null @@ -1,44 +0,0 @@ -import rsa -import requests -from cryptography.fernet import Fernet - -from cmd_chat.client.core.abs.abs_crypto import CryptoService - - -class RSAService(CryptoService): - def __init__(self): - self.public_key = None - self.private_key = None - self.symmetric_key = None - self.fernet = None - self._generate_keys() - - def _encrypt(self, message: str) -> str: - return self.fernet.encrypt(message.encode()).decode("utf-8") - - def _decrypt(self, message: str) -> str: - return self.fernet.decrypt(message.encode()).decode("utf-8") - - def _request_key(self, url: str, username: str, password: str | None = None): - pubkey_bytes = self.public_key.save_pkcs1() - r = requests.post( - url, - files={"pubkey": ("public.pem", pubkey_bytes)}, - data={"username": username, "password": password or ""}, - stream=True, - ) - r.raise_for_status() - # read the full response content (server returns encrypted symmetric key) - message = r.content - self.symmetric_key = rsa.decrypt(message, self.private_key) - self.fernet = Fernet(self.symmetric_key) - - def _generate_keys(self): - self.public_key, self.private_key = rsa.newkeys(512) - - def _get_generated_keys(self): - return self.private_key, self.public_key - - def _remove_keys(self): - self.public_key = None - self.private_key = None \ No newline at end of file diff --git a/.history/cmd_chat/client/core/default_renderer_20251105190717.py b/.history/cmd_chat/client/core/default_renderer_20251105190717.py deleted file mode 100644 index 45252b8..0000000 --- a/.history/cmd_chat/client/core/default_renderer_20251105190717.py +++ /dev/null @@ -1,61 +0,0 @@ -import os -import platform - -from cmd_chat.client.core.abs.abs_renderer import ClientRenderer -from cmd_chat.client.config import COLORS - -from colorama import init - -init() - - -class DefaultClientRenderer(ClientRenderer): - - def __get_os(self) -> str: - """ checking what kind of platform you need - """ - if "Linux" in str(platform.platform()): - return "Linux" - return "Windows" - - def print_message(self, message: str) -> str: - """ generating string with message in required format - """ - # split only on the first ':' to keep message contents intact - message = message.split(":", 1) - if message[0] == self.username: - return COLORS["my_username_color"] + message[0] + ": " + message[1] + COLORS["text_color"] - return message[0] + ": " + message[1] + COLORS["text_color"] - - def clear_console(self): - # For windows clear command its cls - # For linux clear command its clear - if self.__get_os() == "Linux": - os.system("clear") - else: - os.system("cls") - - def print_ip( - self, - ip: str - ) -> str: - return f"IP: " + COLORS["ip_color"] + ip + COLORS["text_color"] - - def print_username( - self, - username: str - ) -> str: - # Username label + colored username - return f"USERNAME: " + COLORS["username_color"] + username + COLORS["text_color"] - - def print_chat(self, response: list[str]) -> str: - for i, msg in enumerate(response["messages"]): - actual_message = self._decrypt(msg) - if i == 0: - for user in response["users_in_chat"]: - print(self.print_ip(user.split(",")[0])) - print(self.print_username(user.split(",")[1])) - print("Write 'q' to quit from chat") - print(f"\n{self.print_message(actual_message)}") - else: - print(f"{self.print_message(actual_message)}") diff --git a/.history/cmd_chat/client/core/default_renderer_20251105190719.py b/.history/cmd_chat/client/core/default_renderer_20251105190719.py deleted file mode 100644 index 45252b8..0000000 --- a/.history/cmd_chat/client/core/default_renderer_20251105190719.py +++ /dev/null @@ -1,61 +0,0 @@ -import os -import platform - -from cmd_chat.client.core.abs.abs_renderer import ClientRenderer -from cmd_chat.client.config import COLORS - -from colorama import init - -init() - - -class DefaultClientRenderer(ClientRenderer): - - def __get_os(self) -> str: - """ checking what kind of platform you need - """ - if "Linux" in str(platform.platform()): - return "Linux" - return "Windows" - - def print_message(self, message: str) -> str: - """ generating string with message in required format - """ - # split only on the first ':' to keep message contents intact - message = message.split(":", 1) - if message[0] == self.username: - return COLORS["my_username_color"] + message[0] + ": " + message[1] + COLORS["text_color"] - return message[0] + ": " + message[1] + COLORS["text_color"] - - def clear_console(self): - # For windows clear command its cls - # For linux clear command its clear - if self.__get_os() == "Linux": - os.system("clear") - else: - os.system("cls") - - def print_ip( - self, - ip: str - ) -> str: - return f"IP: " + COLORS["ip_color"] + ip + COLORS["text_color"] - - def print_username( - self, - username: str - ) -> str: - # Username label + colored username - return f"USERNAME: " + COLORS["username_color"] + username + COLORS["text_color"] - - def print_chat(self, response: list[str]) -> str: - for i, msg in enumerate(response["messages"]): - actual_message = self._decrypt(msg) - if i == 0: - for user in response["users_in_chat"]: - print(self.print_ip(user.split(",")[0])) - print(self.print_username(user.split(",")[1])) - print("Write 'q' to quit from chat") - print(f"\n{self.print_message(actual_message)}") - else: - print(f"{self.print_message(actual_message)}") diff --git a/.history/cmd_chat/client/core/default_renderer_20251105191806.py b/.history/cmd_chat/client/core/default_renderer_20251105191806.py deleted file mode 100644 index b756fb4..0000000 --- a/.history/cmd_chat/client/core/default_renderer_20251105191806.py +++ /dev/null @@ -1,61 +0,0 @@ -import os -import platform - -from cmd_chat.client.core.abs.abs_renderer import ClientRenderer -from cmd_chat.client.config import COLORS - -from colorama import init - -init() - - -class DefaultClientRenderer(ClientRenderer): - - def __get_os(self) -> str: - """ checking what kind of platform you need - """ - if "Linux" in str(platform.platform()): - return "Linux" - return "Windows" - - def print_message(self, message: str) -> str: - """ generating string with message in required format - """ - # split only on the first ':' to keep message contents intact - parts = message.split(":", 1) - if parts[0] == self.username: - return COLORS["my_username_color"] + parts[0] + ": " + parts[1] + COLORS["text_color"] - return parts[0] + ": " + parts[1] + COLORS["text_color"] - - def clear_console(self): - # For windows clear command its cls - # For linux clear command its clear - if self.__get_os() == "Linux": - os.system("clear") - else: - os.system("cls") - - def print_ip( - self, - ip: str - ) -> str: - return f"IP: " + COLORS["ip_color"] + ip + COLORS["text_color"] - - def print_username( - self, - username: str - ) -> str: - # Username label + colored username - return f"USERNAME: " + COLORS["username_color"] + username + COLORS["text_color"] - - def print_chat(self, response: list[str]) -> str: - for i, msg in enumerate(response["messages"]): - actual_message = self._decrypt(msg) - if i == 0: - for user in response["users_in_chat"]: - print(self.print_ip(user.split(",")[0])) - print(self.print_username(user.split(",")[1])) - print("Write 'q' to quit from chat") - print(f"\n{self.print_message(actual_message)}") - else: - print(f"{self.print_message(actual_message)}") diff --git a/.history/cmd_chat/client/core/default_renderer_20251105191832.py b/.history/cmd_chat/client/core/default_renderer_20251105191832.py deleted file mode 100644 index 765bdee..0000000 --- a/.history/cmd_chat/client/core/default_renderer_20251105191832.py +++ /dev/null @@ -1,61 +0,0 @@ -import os -import platform - -from cmd_chat.client.core.abs.abs_renderer import ClientRenderer -from cmd_chat.client.config import COLORS - -from colorama import init - -init() - - -class DefaultClientRenderer(ClientRenderer): - - def __get_os(self) -> str: - """ checking what kind of platform you need - """ - if "Linux" in str(platform.platform()): - return "Linux" - return "Windows" - - def print_message(self, message: str) -> str: - """ generating string with message in required format - """ - # split only on the first ':' to keep message contents intact - parts = message.split(":", 1) - if parts[0] == self.username: - return COLORS["my_username_color"] + parts[0] + ": " + parts[1] + COLORS["text_color"] - return parts[0] + ": " + parts[1] + COLORS["text_color"] - - def clear_console(self): - # For windows clear command its cls - # For linux clear command its clear - if self.__get_os() == "Linux": - os.system("clear") - else: - os.system("cls") - - def print_ip( - self, - ip: str - ) -> str: - return f"IP: " + COLORS["ip_color"] + ip + COLORS["text_color"] - - def print_username( - self, - username: str - ) -> str: - # Username label + colored username - return f"USERNAME: " + COLORS["username_color"] + username + COLORS["text_color"] - - def print_chat(self, response) -> None: - for i, msg in enumerate(response["messages"]): - actual_message = self._decrypt(msg) - if i == 0: - for user in response["users_in_chat"]: - print(self.print_ip(user.split(",")[0])) - print(self.print_username(user.split(",")[1])) - print("Write 'q' to quit from chat") - print(f"\n{self.print_message(actual_message)}") - else: - print(f"{self.print_message(actual_message)}") diff --git a/.history/cmd_chat/client/core/default_renderer_20251105191906.py b/.history/cmd_chat/client/core/default_renderer_20251105191906.py deleted file mode 100644 index 765bdee..0000000 --- a/.history/cmd_chat/client/core/default_renderer_20251105191906.py +++ /dev/null @@ -1,61 +0,0 @@ -import os -import platform - -from cmd_chat.client.core.abs.abs_renderer import ClientRenderer -from cmd_chat.client.config import COLORS - -from colorama import init - -init() - - -class DefaultClientRenderer(ClientRenderer): - - def __get_os(self) -> str: - """ checking what kind of platform you need - """ - if "Linux" in str(platform.platform()): - return "Linux" - return "Windows" - - def print_message(self, message: str) -> str: - """ generating string with message in required format - """ - # split only on the first ':' to keep message contents intact - parts = message.split(":", 1) - if parts[0] == self.username: - return COLORS["my_username_color"] + parts[0] + ": " + parts[1] + COLORS["text_color"] - return parts[0] + ": " + parts[1] + COLORS["text_color"] - - def clear_console(self): - # For windows clear command its cls - # For linux clear command its clear - if self.__get_os() == "Linux": - os.system("clear") - else: - os.system("cls") - - def print_ip( - self, - ip: str - ) -> str: - return f"IP: " + COLORS["ip_color"] + ip + COLORS["text_color"] - - def print_username( - self, - username: str - ) -> str: - # Username label + colored username - return f"USERNAME: " + COLORS["username_color"] + username + COLORS["text_color"] - - def print_chat(self, response) -> None: - for i, msg in enumerate(response["messages"]): - actual_message = self._decrypt(msg) - if i == 0: - for user in response["users_in_chat"]: - print(self.print_ip(user.split(",")[0])) - print(self.print_username(user.split(",")[1])) - print("Write 'q' to quit from chat") - print(f"\n{self.print_message(actual_message)}") - else: - print(f"{self.print_message(actual_message)}") diff --git a/.history/cmd_chat/client/core/rich_renderer_20251105190741.py b/.history/cmd_chat/client/core/rich_renderer_20251105190741.py deleted file mode 100644 index 733f540..0000000 --- a/.history/cmd_chat/client/core/rich_renderer_20251105190741.py +++ /dev/null @@ -1,78 +0,0 @@ -import os -import platform - -from rich.text import Text -from rich.style import Style -from rich.console import Console - -from rich.table import Table -from cmd_chat.client.core.abs.abs_renderer import ClientRenderer -from cmd_chat.client.config import MESSAGES_TO_SHOW - - -console = Console(width=75) - - -class RichClientRenderer(ClientRenderer): - - def __get_os(self) -> str: - """ checking what kind of platform you need - """ - if "Linux" in str(platform.platform()): - return "Linux" - return "Windows" - - def print_message(self, message: str) -> Text: - """ generating string with message in required format - """ - # split only on the first ':' so message bodies containing ':' are preserved - message = message.split(":", 1) - if message[0] == self.username: - return \ - Text(text=message[0], style="bold") + \ - Text(text=": ", style="bold") + \ - Text(text=message[1], style="underline") - return \ - Text(text=message[0], style="bold") + \ - Text(text=": ", style="bold") + \ - Text(text=message[1], style="underline") - - def clear_console(self): - # For windows clear command its cls - # For linux clear command its clear - if self.__get_os() == "Linux": - os.system("clear") - else: - os.system("cls") - - def print_ip( - self, - ip: str - ) -> str: - return ip - - def print_username( - self, - username: str - ) -> str: - return username - - def print_chat(self, response: list[str]) -> str: - self.clear_console() - for i, msg in enumerate(response["messages"][-MESSAGES_TO_SHOW:]): - actual_message = self._decrypt(msg) - if i == 0: - console.print("Users in chat:", justify="left") - table = Table(show_header=True, header_style="bold magenta") - table.add_column("IP", style="dim", width=12) - table.add_column("USERNAME") - for user in response["users_in_chat"]: - table.add_row( - self.print_ip(user.split(',')[0]), - self.print_username(user.split(",")[1]) - ) - console.print(table) - console.print("Write 'q' to quit from chat", justify="left") - console.print(f"\n{self.print_message(actual_message)}") - else: - console.print(f"{self.print_message(actual_message)}") \ No newline at end of file diff --git a/.history/cmd_chat/client/core/rich_renderer_20251105190758.py b/.history/cmd_chat/client/core/rich_renderer_20251105190758.py deleted file mode 100644 index 733f540..0000000 --- a/.history/cmd_chat/client/core/rich_renderer_20251105190758.py +++ /dev/null @@ -1,78 +0,0 @@ -import os -import platform - -from rich.text import Text -from rich.style import Style -from rich.console import Console - -from rich.table import Table -from cmd_chat.client.core.abs.abs_renderer import ClientRenderer -from cmd_chat.client.config import MESSAGES_TO_SHOW - - -console = Console(width=75) - - -class RichClientRenderer(ClientRenderer): - - def __get_os(self) -> str: - """ checking what kind of platform you need - """ - if "Linux" in str(platform.platform()): - return "Linux" - return "Windows" - - def print_message(self, message: str) -> Text: - """ generating string with message in required format - """ - # split only on the first ':' so message bodies containing ':' are preserved - message = message.split(":", 1) - if message[0] == self.username: - return \ - Text(text=message[0], style="bold") + \ - Text(text=": ", style="bold") + \ - Text(text=message[1], style="underline") - return \ - Text(text=message[0], style="bold") + \ - Text(text=": ", style="bold") + \ - Text(text=message[1], style="underline") - - def clear_console(self): - # For windows clear command its cls - # For linux clear command its clear - if self.__get_os() == "Linux": - os.system("clear") - else: - os.system("cls") - - def print_ip( - self, - ip: str - ) -> str: - return ip - - def print_username( - self, - username: str - ) -> str: - return username - - def print_chat(self, response: list[str]) -> str: - self.clear_console() - for i, msg in enumerate(response["messages"][-MESSAGES_TO_SHOW:]): - actual_message = self._decrypt(msg) - if i == 0: - console.print("Users in chat:", justify="left") - table = Table(show_header=True, header_style="bold magenta") - table.add_column("IP", style="dim", width=12) - table.add_column("USERNAME") - for user in response["users_in_chat"]: - table.add_row( - self.print_ip(user.split(',')[0]), - self.print_username(user.split(",")[1]) - ) - console.print(table) - console.print("Write 'q' to quit from chat", justify="left") - console.print(f"\n{self.print_message(actual_message)}") - else: - console.print(f"{self.print_message(actual_message)}") \ No newline at end of file diff --git a/.history/cmd_chat/client/core/rich_renderer_20251105191843.py b/.history/cmd_chat/client/core/rich_renderer_20251105191843.py deleted file mode 100644 index 30efbf8..0000000 --- a/.history/cmd_chat/client/core/rich_renderer_20251105191843.py +++ /dev/null @@ -1,78 +0,0 @@ -import os -import platform - -from rich.text import Text -from rich.style import Style -from rich.console import Console - -from rich.table import Table -from cmd_chat.client.core.abs.abs_renderer import ClientRenderer -from cmd_chat.client.config import MESSAGES_TO_SHOW - - -console = Console(width=75) - - -class RichClientRenderer(ClientRenderer): - - def __get_os(self) -> str: - """ checking what kind of platform you need - """ - if "Linux" in str(platform.platform()): - return "Linux" - return "Windows" - - def print_message(self, message: str) -> Text: - """ generating string with message in required format - """ - # split only on the first ':' so message bodies containing ':' are preserved - parts = message.split(":", 1) - if parts[0] == self.username: - return \ - Text(text=parts[0], style="bold") + \ - Text(text=": ", style="bold") + \ - Text(text=parts[1], style="underline") - return \ - Text(text=parts[0], style="bold") + \ - Text(text=": ", style="bold") + \ - Text(text=parts[1], style="underline") - - def clear_console(self): - # For windows clear command its cls - # For linux clear command its clear - if self.__get_os() == "Linux": - os.system("clear") - else: - os.system("cls") - - def print_ip( - self, - ip: str - ) -> str: - return ip - - def print_username( - self, - username: str - ) -> str: - return username - - def print_chat(self, response: list[str]) -> str: - self.clear_console() - for i, msg in enumerate(response["messages"][-MESSAGES_TO_SHOW:]): - actual_message = self._decrypt(msg) - if i == 0: - console.print("Users in chat:", justify="left") - table = Table(show_header=True, header_style="bold magenta") - table.add_column("IP", style="dim", width=12) - table.add_column("USERNAME") - for user in response["users_in_chat"]: - table.add_row( - self.print_ip(user.split(',')[0]), - self.print_username(user.split(",")[1]) - ) - console.print(table) - console.print("Write 'q' to quit from chat", justify="left") - console.print(f"\n{self.print_message(actual_message)}") - else: - console.print(f"{self.print_message(actual_message)}") \ No newline at end of file diff --git a/.history/cmd_chat/client/core/rich_renderer_20251105191852.py b/.history/cmd_chat/client/core/rich_renderer_20251105191852.py deleted file mode 100644 index 0f40c3c..0000000 --- a/.history/cmd_chat/client/core/rich_renderer_20251105191852.py +++ /dev/null @@ -1,78 +0,0 @@ -import os -import platform - -from rich.text import Text -from rich.style import Style -from rich.console import Console - -from rich.table import Table -from cmd_chat.client.core.abs.abs_renderer import ClientRenderer -from cmd_chat.client.config import MESSAGES_TO_SHOW - - -console = Console(width=75) - - -class RichClientRenderer(ClientRenderer): - - def __get_os(self) -> str: - """ checking what kind of platform you need - """ - if "Linux" in str(platform.platform()): - return "Linux" - return "Windows" - - def print_message(self, message: str) -> Text: - """ generating string with message in required format - """ - # split only on the first ':' so message bodies containing ':' are preserved - parts = message.split(":", 1) - if parts[0] == self.username: - return \ - Text(text=parts[0], style="bold") + \ - Text(text=": ", style="bold") + \ - Text(text=parts[1], style="underline") - return \ - Text(text=parts[0], style="bold") + \ - Text(text=": ", style="bold") + \ - Text(text=parts[1], style="underline") - - def clear_console(self): - # For windows clear command its cls - # For linux clear command its clear - if self.__get_os() == "Linux": - os.system("clear") - else: - os.system("cls") - - def print_ip( - self, - ip: str - ) -> str: - return ip - - def print_username( - self, - username: str - ) -> str: - return username - - def print_chat(self, response) -> None: - self.clear_console() - for i, msg in enumerate(response["messages"][-MESSAGES_TO_SHOW:]): - actual_message = self._decrypt(msg) - if i == 0: - console.print("Users in chat:", justify="left") - table = Table(show_header=True, header_style="bold magenta") - table.add_column("IP", style="dim", width=12) - table.add_column("USERNAME") - for user in response["users_in_chat"]: - table.add_row( - self.print_ip(user.split(',')[0]), - self.print_username(user.split(",")[1]) - ) - console.print(table) - console.print("Write 'q' to quit from chat", justify="left") - console.print(f"\n{self.print_message(actual_message)}") - else: - console.print(f"{self.print_message(actual_message)}") \ No newline at end of file diff --git a/.history/cmd_chat/client/core/rich_renderer_20251105191906.py b/.history/cmd_chat/client/core/rich_renderer_20251105191906.py deleted file mode 100644 index 0f40c3c..0000000 --- a/.history/cmd_chat/client/core/rich_renderer_20251105191906.py +++ /dev/null @@ -1,78 +0,0 @@ -import os -import platform - -from rich.text import Text -from rich.style import Style -from rich.console import Console - -from rich.table import Table -from cmd_chat.client.core.abs.abs_renderer import ClientRenderer -from cmd_chat.client.config import MESSAGES_TO_SHOW - - -console = Console(width=75) - - -class RichClientRenderer(ClientRenderer): - - def __get_os(self) -> str: - """ checking what kind of platform you need - """ - if "Linux" in str(platform.platform()): - return "Linux" - return "Windows" - - def print_message(self, message: str) -> Text: - """ generating string with message in required format - """ - # split only on the first ':' so message bodies containing ':' are preserved - parts = message.split(":", 1) - if parts[0] == self.username: - return \ - Text(text=parts[0], style="bold") + \ - Text(text=": ", style="bold") + \ - Text(text=parts[1], style="underline") - return \ - Text(text=parts[0], style="bold") + \ - Text(text=": ", style="bold") + \ - Text(text=parts[1], style="underline") - - def clear_console(self): - # For windows clear command its cls - # For linux clear command its clear - if self.__get_os() == "Linux": - os.system("clear") - else: - os.system("cls") - - def print_ip( - self, - ip: str - ) -> str: - return ip - - def print_username( - self, - username: str - ) -> str: - return username - - def print_chat(self, response) -> None: - self.clear_console() - for i, msg in enumerate(response["messages"][-MESSAGES_TO_SHOW:]): - actual_message = self._decrypt(msg) - if i == 0: - console.print("Users in chat:", justify="left") - table = Table(show_header=True, header_style="bold magenta") - table.add_column("IP", style="dim", width=12) - table.add_column("USERNAME") - for user in response["users_in_chat"]: - table.add_row( - self.print_ip(user.split(',')[0]), - self.print_username(user.split(",")[1]) - ) - console.print(table) - console.print("Write 'q' to quit from chat", justify="left") - console.print(f"\n{self.print_message(actual_message)}") - else: - console.print(f"{self.print_message(actual_message)}") \ No newline at end of file diff --git a/.history/cmd_chat/server/server_20251105191755.py b/.history/cmd_chat/server/server_20251105191755.py deleted file mode 100644 index 33bbe36..0000000 --- a/.history/cmd_chat/server/server_20251105191755.py +++ /dev/null @@ -1,113 +0,0 @@ -import asyncio -import rsa -from cryptography.fernet import Fernet -from functools import partial -from sanic.worker.loader import AppLoader -from sanic.response import HTTPResponse -from sanic import Sanic, Request, response, Websocket -from cmd_chat.server.models import Message -from cmd_chat.server.services import ( - _get_bytes_and_serialize, - _check_ws_for_close_status, - _generate_new_message, - _generate_update_payload -) - -app = Sanic("app") -app.config.OAS = False - -MESSAGES_MEMORY_DB: list[Message] = [] -USERS: dict[str, str] = {} -PUBLIC_KEY = Fernet.generate_key() - - -def _check_password(request: Request, expected: str | None) -> bool: - if not expected: - return True - q = request.args.get("password") - f = request.form.get("password") if hasattr(request, "form") else None - return (q or f) == expected - -def _get_str_arg(request: Request, name: str) -> str | None: - return request.form.get(name) or request.args.get(name) - -def attach_endpoints(app: Sanic): - @app.websocket("/talk") - async def talk_ws_view(request: Request, ws: Websocket) -> HTTPResponse: - if not _check_password(request, app.ctx.ADMIN_PASSWORD): - await ws.close(code=4001, reason="unauthorized") - return - while True: - serialized_message: dict = await _get_bytes_and_serialize(ws) - await _check_ws_for_close_status(serialized_message, ws) - text = serialized_message.get("text") - if text is None: - continue - new_message = await _generate_new_message(text) - MESSAGES_MEMORY_DB.append(new_message) - await ws.send(str({"status": "ok"})) - await asyncio.sleep(0.2) - - @app.websocket("/update") - async def update_ws_view(request: Request, ws: Websocket) -> HTTPResponse: - if not _check_password(request, app.ctx.ADMIN_PASSWORD): - await ws.close(code=4001, reason="unauthorized") - return - while True: - payload = await _generate_update_payload(MESSAGES_MEMORY_DB, USERS) - await ws.send(payload.encode()) - await asyncio.sleep(0.2) - - @app.route('/get_key', methods=['GET', 'POST']) - async def get_key_view(request: Request) -> HTTPResponse: - if not _check_password(request, app.ctx.ADMIN_PASSWORD): - return response.text("unauthorized", status=401) - - pubkey_bytes: bytes | None = None - - if "pubkey" in request.files and request.files.get("pubkey"): - f = request.files.get("pubkey") - if isinstance(f, list): - f = f[0] - pubkey_bytes = f.body - - if pubkey_bytes is None: - raw = request.form.get("pubkey") - if raw: - pubkey_bytes = raw if isinstance(raw, bytes) else str(raw).encode() - - if pubkey_bytes is None: - raw = request.args.get("pubkey") - if raw: - pubkey_bytes = raw.encode() - - if not pubkey_bytes: - return response.text("bad request: pubkey is required", status=400) - - try: - public_key = rsa.PublicKey.load_pkcs1(pubkey_bytes) - except Exception as e: - return response.text(f"bad pubkey: {e}", status=400) - - encrypted_data = rsa.encrypt(PUBLIC_KEY, public_key) - - username = _get_str_arg(request, "username") or "unknown" - user_key = f"{request.ip}, {username}" - if user_key not in USERS: - USERS[user_key] = PUBLIC_KEY - - return response.raw(encrypted_data) - - -def create_app(app_name: str, admin_password: str | None) -> Sanic: - app = Sanic(app_name) - app.ctx.ADMIN_PASSWORD = admin_password - attach_endpoints(app) - return app - - -def run_server(host: str, port: int, dev: bool = False, admin_password: str | None = None) -> None: - loader = AppLoader(factory=partial(create_app, "CMD_SERVER", admin_password)) - app = loader.load() - app.prepare(host=host, port=port, dev=dev) - Sanic.serve(primary=app, app_loader=loader) diff --git a/.history/cmd_chat/server/server_20251105191907.py b/.history/cmd_chat/server/server_20251105191907.py deleted file mode 100644 index 33bbe36..0000000 --- a/.history/cmd_chat/server/server_20251105191907.py +++ /dev/null @@ -1,113 +0,0 @@ -import asyncio -import rsa -from cryptography.fernet import Fernet -from functools import partial -from sanic.worker.loader import AppLoader -from sanic.response import HTTPResponse -from sanic import Sanic, Request, response, Websocket -from cmd_chat.server.models import Message -from cmd_chat.server.services import ( - _get_bytes_and_serialize, - _check_ws_for_close_status, - _generate_new_message, - _generate_update_payload -) - -app = Sanic("app") -app.config.OAS = False - -MESSAGES_MEMORY_DB: list[Message] = [] -USERS: dict[str, str] = {} -PUBLIC_KEY = Fernet.generate_key() - - -def _check_password(request: Request, expected: str | None) -> bool: - if not expected: - return True - q = request.args.get("password") - f = request.form.get("password") if hasattr(request, "form") else None - return (q or f) == expected - -def _get_str_arg(request: Request, name: str) -> str | None: - return request.form.get(name) or request.args.get(name) - -def attach_endpoints(app: Sanic): - @app.websocket("/talk") - async def talk_ws_view(request: Request, ws: Websocket) -> HTTPResponse: - if not _check_password(request, app.ctx.ADMIN_PASSWORD): - await ws.close(code=4001, reason="unauthorized") - return - while True: - serialized_message: dict = await _get_bytes_and_serialize(ws) - await _check_ws_for_close_status(serialized_message, ws) - text = serialized_message.get("text") - if text is None: - continue - new_message = await _generate_new_message(text) - MESSAGES_MEMORY_DB.append(new_message) - await ws.send(str({"status": "ok"})) - await asyncio.sleep(0.2) - - @app.websocket("/update") - async def update_ws_view(request: Request, ws: Websocket) -> HTTPResponse: - if not _check_password(request, app.ctx.ADMIN_PASSWORD): - await ws.close(code=4001, reason="unauthorized") - return - while True: - payload = await _generate_update_payload(MESSAGES_MEMORY_DB, USERS) - await ws.send(payload.encode()) - await asyncio.sleep(0.2) - - @app.route('/get_key', methods=['GET', 'POST']) - async def get_key_view(request: Request) -> HTTPResponse: - if not _check_password(request, app.ctx.ADMIN_PASSWORD): - return response.text("unauthorized", status=401) - - pubkey_bytes: bytes | None = None - - if "pubkey" in request.files and request.files.get("pubkey"): - f = request.files.get("pubkey") - if isinstance(f, list): - f = f[0] - pubkey_bytes = f.body - - if pubkey_bytes is None: - raw = request.form.get("pubkey") - if raw: - pubkey_bytes = raw if isinstance(raw, bytes) else str(raw).encode() - - if pubkey_bytes is None: - raw = request.args.get("pubkey") - if raw: - pubkey_bytes = raw.encode() - - if not pubkey_bytes: - return response.text("bad request: pubkey is required", status=400) - - try: - public_key = rsa.PublicKey.load_pkcs1(pubkey_bytes) - except Exception as e: - return response.text(f"bad pubkey: {e}", status=400) - - encrypted_data = rsa.encrypt(PUBLIC_KEY, public_key) - - username = _get_str_arg(request, "username") or "unknown" - user_key = f"{request.ip}, {username}" - if user_key not in USERS: - USERS[user_key] = PUBLIC_KEY - - return response.raw(encrypted_data) - - -def create_app(app_name: str, admin_password: str | None) -> Sanic: - app = Sanic(app_name) - app.ctx.ADMIN_PASSWORD = admin_password - attach_endpoints(app) - return app - - -def run_server(host: str, port: int, dev: bool = False, admin_password: str | None = None) -> None: - loader = AppLoader(factory=partial(create_app, "CMD_SERVER", admin_password)) - app = loader.load() - app.prepare(host=host, port=port, dev=dev) - Sanic.serve(primary=app, app_loader=loader) diff --git a/.history/cmd_chat/server/services_20251105191745.py b/.history/cmd_chat/server/services_20251105191745.py deleted file mode 100644 index e951d32..0000000 --- a/.history/cmd_chat/server/services_20251105191745.py +++ /dev/null @@ -1,36 +0,0 @@ -import ast -from sanic import Websocket -from cmd_chat.server.models import Message - - -async def _get_bytes_and_serialize( - ws: Websocket -) -> dict: - return ast.literal_eval(await ws.recv()) - - -async def _check_ws_for_close_status( - response: dict, - ws: Websocket -) -> None: - if "action" in response.keys(): - if response["action"] == "close": - await ws.close() - - -async def _generate_new_message( - message: str -) -> Message: - return Message(message = message) - - -async def _generate_update_payload( - memory_msgs: list[Message], - users_structure: dict -) -> str: - return str({ - "messages": [i.message for i in memory_msgs], - "users_in_chat": list(users_structure.keys()) - }) - - diff --git a/.history/cmd_chat/server/services_20251105191746.py b/.history/cmd_chat/server/services_20251105191746.py deleted file mode 100644 index e951d32..0000000 --- a/.history/cmd_chat/server/services_20251105191746.py +++ /dev/null @@ -1,36 +0,0 @@ -import ast -from sanic import Websocket -from cmd_chat.server.models import Message - - -async def _get_bytes_and_serialize( - ws: Websocket -) -> dict: - return ast.literal_eval(await ws.recv()) - - -async def _check_ws_for_close_status( - response: dict, - ws: Websocket -) -> None: - if "action" in response.keys(): - if response["action"] == "close": - await ws.close() - - -async def _generate_new_message( - message: str -) -> Message: - return Message(message = message) - - -async def _generate_update_payload( - memory_msgs: list[Message], - users_structure: dict -) -> str: - return str({ - "messages": [i.message for i in memory_msgs], - "users_in_chat": list(users_structure.keys()) - }) - - diff --git a/.history/setup_20251105190626.py b/.history/setup_20251105190626.py deleted file mode 100644 index 7eef936..0000000 --- a/.history/setup_20251105190626.py +++ /dev/null @@ -1,33 +0,0 @@ -import setuptools - -with open("README.md", "r", encoding="utf-8") as fh: - description = fh.read() - -setuptools.setup( - name="secured_console_chat", - version="1.1.22", - author="dinosaurtirex", - author_email="sneakybeaky18@gmail.com", - # Use find_packages to correctly discover package names - packages=setuptools.find_packages(exclude=("tests", "docs")), - description="Secured console chat with RSA & Fernet", - long_description=description, - long_description_content_type="text/markdown", - url="https://github.com/dinosaurtirex/cmd-chat", - license='MIT', - python_requires='>=3.10', - entry_points={ - 'console_scripts': [ - 'cmd_chat = cmd_chat:main' - ] - }, - install_requires=[ - "sanic", - "requests", - "rsa", - "cryptography", - "colorama", - "pydantic", - "websocket-client" - ] -) \ No newline at end of file diff --git a/.history/setup_20251105190632.py b/.history/setup_20251105190632.py deleted file mode 100644 index 7eef936..0000000 --- a/.history/setup_20251105190632.py +++ /dev/null @@ -1,33 +0,0 @@ -import setuptools - -with open("README.md", "r", encoding="utf-8") as fh: - description = fh.read() - -setuptools.setup( - name="secured_console_chat", - version="1.1.22", - author="dinosaurtirex", - author_email="sneakybeaky18@gmail.com", - # Use find_packages to correctly discover package names - packages=setuptools.find_packages(exclude=("tests", "docs")), - description="Secured console chat with RSA & Fernet", - long_description=description, - long_description_content_type="text/markdown", - url="https://github.com/dinosaurtirex/cmd-chat", - license='MIT', - python_requires='>=3.10', - entry_points={ - 'console_scripts': [ - 'cmd_chat = cmd_chat:main' - ] - }, - install_requires=[ - "sanic", - "requests", - "rsa", - "cryptography", - "colorama", - "pydantic", - "websocket-client" - ] -) \ No newline at end of file