154 lines
5.2 KiB
Python
154 lines
5.2 KiB
Python
from __future__ import annotations
|
|
|
|
import queue
|
|
import threading
|
|
from typing import Optional
|
|
|
|
from secure_sms.infrastructure.database import Database
|
|
from secure_sms.infrastructure.gsm import IMessageGateway, GSMGateway
|
|
from secure_sms.application.services import SecureMessagingService
|
|
|
|
|
|
class AppController:
|
|
def __init__(self):
|
|
self.db = Database()
|
|
self.service = SecureMessagingService(self.db)
|
|
self.gsm: Optional[IMessageGateway] = None
|
|
self.ui = None
|
|
self._outbox = queue.Queue()
|
|
self._worker_thread = threading.Thread(target=self._outbox_worker, daemon=True)
|
|
self._worker_thread.start()
|
|
|
|
def _outbox_worker(self):
|
|
while True:
|
|
try:
|
|
task = self._outbox.get()
|
|
if task is None:
|
|
break
|
|
phone = task.get("phone")
|
|
frames = task.get("frames")
|
|
msg_id = task.get("msg_id")
|
|
|
|
if self.gsm and self.gsm.is_connected:
|
|
sent = self.gsm.send_frames(phone, frames)
|
|
state = "sent" if sent else "failed"
|
|
else:
|
|
# In simulation mode (no modem attached), we instantly mark it simul-sent.
|
|
sent = True
|
|
state = "simulated"
|
|
|
|
if msg_id:
|
|
self.db.update_message_transport_state(msg_id, state)
|
|
self._notify_ui(phone)
|
|
self._outbox.task_done()
|
|
except Exception:
|
|
pass
|
|
|
|
def bind_ui(self, ui):
|
|
self.ui = ui
|
|
|
|
def is_bootstrapped(self) -> bool:
|
|
return self.service.is_bootstrapped()
|
|
|
|
def bootstrap(self, password: str):
|
|
self.service.bootstrap(password)
|
|
self._start_gsm()
|
|
|
|
def unlock(self, password: str) -> bool:
|
|
unlocked = self.service.unlock(password)
|
|
if unlocked:
|
|
self._start_gsm()
|
|
return unlocked
|
|
|
|
def verify_password(self, password: str) -> bool:
|
|
return self.service.verify_password(password)
|
|
|
|
def change_master_password(self, current_password: str, new_password: str):
|
|
self.service.change_master_password(current_password, new_password)
|
|
|
|
def _start_gsm(self) -> bool:
|
|
self._stop_gsm()
|
|
port, baudrate = self.service.get_connection_settings()
|
|
self.gsm = GSMGateway(port=port, baudrate=baudrate, message_callback=self._on_sms_received)
|
|
connected = self.gsm.connect()
|
|
self._notify_ui()
|
|
return connected
|
|
|
|
def reconnect_modem(self, port: str, baudrate: int) -> bool:
|
|
self.service.update_connection_settings(port, baudrate)
|
|
return self._start_gsm()
|
|
|
|
def modem_status(self) -> dict:
|
|
return {
|
|
"connected": bool(self.gsm and self.gsm.is_connected),
|
|
"port": self.gsm.port if self.gsm else self.service.get_connection_settings()[0],
|
|
"baudrate": self.gsm.baudrate if self.gsm else self.service.get_connection_settings()[1],
|
|
}
|
|
|
|
def list_contacts(self):
|
|
return self.service.list_contacts()
|
|
|
|
def get_contact(self, phone: str):
|
|
return self.service.get_contact(phone)
|
|
|
|
def save_contact(self, name: str, phone: str):
|
|
self.service.add_or_update_contact(name, phone)
|
|
self._notify_ui()
|
|
|
|
def delete_contact(self, phone: str):
|
|
self.service.delete_contact(phone)
|
|
self._notify_ui()
|
|
|
|
def get_messages(self, phone: str):
|
|
return self.service.get_messages(phone)
|
|
|
|
def send_message(self, phone: str, text: str) -> tuple[bool, str]:
|
|
try:
|
|
frames, mode = self.service.prepare_outgoing_message(phone, text)
|
|
except Exception as exc:
|
|
return False, str(exc)
|
|
|
|
msg_id = self.service.store_outgoing_message(phone, text, mode, "queued")
|
|
self._outbox.put({"phone": phone, "frames": frames, "msg_id": msg_id})
|
|
self._notify_ui(phone)
|
|
return True, "queued"
|
|
|
|
def request_secure(self, phone: str) -> tuple[bool, str]:
|
|
frames = self.service.request_secure_channel(phone)
|
|
self._outbox.put({"phone": phone, "frames": frames, "msg_id": None})
|
|
self._notify_ui(phone)
|
|
return True, "queued"
|
|
|
|
def switch_to_normal(self, phone: str) -> tuple[bool, str]:
|
|
frames = self.service.request_normal_mode(phone)
|
|
self._outbox.put({"phone": phone, "frames": frames, "msg_id": None})
|
|
self._notify_ui(phone)
|
|
return True, "queued"
|
|
|
|
def get_admin_snapshot(self) -> dict:
|
|
snapshot = self.service.get_admin_snapshot()
|
|
snapshot["modem"] = self.modem_status()
|
|
return snapshot
|
|
|
|
def _on_sms_received(self, sender: str, raw_text: str):
|
|
try:
|
|
_, reply_frames = self.service.process_incoming_sms(sender, raw_text)
|
|
if reply_frames and self.gsm and self.gsm.is_connected:
|
|
self.gsm.send_frames(sender, reply_frames)
|
|
finally:
|
|
self._notify_ui(sender)
|
|
|
|
def _notify_ui(self, phone: Optional[str] = None):
|
|
if self.ui:
|
|
self.ui.after(0, lambda: self.ui.handle_background_refresh(phone))
|
|
|
|
def _stop_gsm(self):
|
|
if self.gsm:
|
|
self.gsm.disconnect()
|
|
self.gsm = None
|
|
|
|
def shutdown(self):
|
|
self._stop_gsm()
|
|
if self.ui:
|
|
self.ui.destroy()
|