Trilltechnician/hack-house
1
2
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

Remove .history folder

Browse Source
This commit is contained in:
mirai 2025-11-07 16:33:18 +05:30
parent 64b0967292
commit 0759518dce
26 changed files with 0 additions and 1645 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

162
.history/cmd_chat/client/client_20251105191919.py
View File

@ -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()

162
.history/cmd_chat/client/client_20251105191921.py
View File

@ -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()

27
.history/cmd_chat/client/core/abs/abs_crypto_20251105190705.py
View File

@ -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")

27
.history/cmd_chat/client/core/abs/abs_crypto_20251105190714.py
View File

@ -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")

26
.history/cmd_chat/client/core/abs/abs_renderer_20251105190654.py
View File

@ -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")

26
.history/cmd_chat/client/core/abs/abs_renderer_20251105190714.py
View File

@ -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")

34
.history/cmd_chat/client/core/abs/abs_renderer_20251105191905.py
View File

@ -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")

34
.history/cmd_chat/client/core/abs/abs_renderer_20251105191906.py
View File

@ -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")

44
.history/cmd_chat/client/core/crypto_20251105190754.py
View File

@ -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

44
.history/cmd_chat/client/core/crypto_20251105190758.py
View File

@ -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

61
.history/cmd_chat/client/core/default_renderer_20251105190717.py
View File

@ -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)}")

61
.history/cmd_chat/client/core/default_renderer_20251105190719.py
View File

@ -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)}")

61
.history/cmd_chat/client/core/default_renderer_20251105191806.py
View File

@ -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)}")

61
.history/cmd_chat/client/core/default_renderer_20251105191832.py
View File

@ -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)}")

61
.history/cmd_chat/client/core/default_renderer_20251105191906.py
View File

@ -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)}")

78
.history/cmd_chat/client/core/rich_renderer_20251105190741.py
View File

@ -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)}")

78
.history/cmd_chat/client/core/rich_renderer_20251105190758.py
View File

@ -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)}")

78
.history/cmd_chat/client/core/rich_renderer_20251105191843.py
View File

@ -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)}")

78
.history/cmd_chat/client/core/rich_renderer_20251105191852.py
View File

@ -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)}")

78
.history/cmd_chat/client/core/rich_renderer_20251105191906.py
View File

@ -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)}")

113
.history/cmd_chat/server/server_20251105191755.py
View File

@ -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)

113
.history/cmd_chat/server/server_20251105191907.py
View File

@ -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)

36
.history/cmd_chat/server/services_20251105191745.py
View File

@ -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())
})

36
.history/cmd_chat/server/services_20251105191746.py
View File

@ -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())
})

33
.history/setup_20251105190626.py
View File

@ -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"
]
)

33
.history/setup_20251105190632.py
View File

@ -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"
]
)