95f8a192b5
Server: - Split into views, routes, helpers, models modules - Merged /ws/talk and /ws/update into single /ws/chat endpoint - Replaced polling with push-based broadcast model - Added username uniqueness validation on connect - Fixed run_server arguments bug (workers parameter) - Removed deprecated loop argument from Sanic listeners - Replaced datetime.utcnow() with timezone-aware datetime.now(timezone.utc) Client: - Rewrote client as single-file module - Migrated from websocket-client to websockets (asyncio) - Fixed websocket-client conflict with asyncio event loop on Windows - Added progress indicators for key generation, exchange, connection - Added animated 3D spinning cube in UI - Updated RSA key from 512 to 2048 bits CLI: - Removed unnecessary asyncio.run() wrapper - Simplified entry point
44 lines
1.3 KiB
Python
44 lines
1.3 KiB
Python
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()
|
|
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(2048)
|
|
|
|
def _get_generated_keys(self):
|
|
return self.private_key, self.public_key
|
|
|
|
def _remove_keys(self):
|
|
self.public_key = None
|
|
self.private_key = None
|