79c2a4d4bf
Replace monolithic CLI and web server with route-based Flask API. New routes: api_commands, api_build, api_can, api_monitor, api_ota, api_tunnel. Add honeypot security dashboard with real-time SSE, MITRE ATT&CK mapping, kill chain analysis. New TUI with commander/help modules. Add session management, tunnel proxy core, CAN bus data store. Docker support.
110 lines
3.2 KiB
Python
110 lines
3.2 KiB
Python
import time
|
|
|
|
# TUI bridge import (lazy to avoid circular imports)
|
|
_tui_bridge = None
|
|
|
|
|
|
def _get_bridge():
|
|
global _tui_bridge
|
|
if _tui_bridge is None:
|
|
try:
|
|
from tui.bridge import tui_bridge
|
|
_tui_bridge = tui_bridge
|
|
except ImportError:
|
|
_tui_bridge = False
|
|
return _tui_bridge if _tui_bridge else None
|
|
|
|
|
|
class Display:
|
|
_tui_mode = False
|
|
|
|
@classmethod
|
|
def enable_tui_mode(cls):
|
|
cls._tui_mode = True
|
|
|
|
@staticmethod
|
|
def _timestamp() -> str:
|
|
return time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
|
|
|
|
@staticmethod
|
|
def system_message(message: str):
|
|
bridge = _get_bridge()
|
|
if bridge:
|
|
from tui.bridge import TUIMessage, MessageType
|
|
bridge.post_message(TUIMessage(
|
|
msg_type=MessageType.SYSTEM_MESSAGE,
|
|
payload=message
|
|
))
|
|
|
|
@staticmethod
|
|
def device_event(device_id: str, event: str):
|
|
bridge = _get_bridge()
|
|
if bridge:
|
|
from tui.bridge import TUIMessage, MessageType
|
|
if "Connected from" in event:
|
|
msg_type = MessageType.DEVICE_CONNECTED
|
|
elif "Reconnected from" in event:
|
|
msg_type = MessageType.DEVICE_RECONNECTED
|
|
elif event == "Disconnected":
|
|
msg_type = MessageType.DEVICE_DISCONNECTED
|
|
else:
|
|
msg_type = MessageType.DEVICE_EVENT
|
|
bridge.post_message(TUIMessage(
|
|
msg_type=msg_type,
|
|
device_id=device_id,
|
|
payload=event
|
|
))
|
|
|
|
@staticmethod
|
|
def command_sent(device_id: str, command_name: str, request_id: str):
|
|
bridge = _get_bridge()
|
|
if bridge:
|
|
from tui.bridge import TUIMessage, MessageType
|
|
bridge.post_message(TUIMessage(
|
|
msg_type=MessageType.COMMAND_SENT,
|
|
device_id=device_id,
|
|
payload=command_name,
|
|
request_id=request_id
|
|
))
|
|
|
|
@staticmethod
|
|
def command_response(request_id: str, device_id: str, response: str):
|
|
bridge = _get_bridge()
|
|
if bridge:
|
|
from tui.bridge import TUIMessage, MessageType
|
|
bridge.post_message(TUIMessage(
|
|
msg_type=MessageType.COMMAND_RESPONSE,
|
|
device_id=device_id,
|
|
payload=response,
|
|
request_id=request_id
|
|
))
|
|
|
|
@staticmethod
|
|
def error(message: str):
|
|
bridge = _get_bridge()
|
|
if bridge:
|
|
from tui.bridge import TUIMessage, MessageType
|
|
bridge.post_message(TUIMessage(
|
|
msg_type=MessageType.ERROR,
|
|
payload=message
|
|
))
|
|
|
|
@staticmethod
|
|
def format_duration(seconds: float) -> str:
|
|
seconds = int(seconds)
|
|
m, s = divmod(seconds, 60)
|
|
h, m = divmod(m, 60)
|
|
d, h = divmod(h, 24)
|
|
|
|
if d > 0:
|
|
return f"{d}d {h}h {m}m"
|
|
if h > 0:
|
|
return f"{h}h {m}m {s}s"
|
|
if m > 0:
|
|
return f"{m}m {s}s"
|
|
return f"{s}s"
|
|
|
|
@staticmethod
|
|
def format_timestamp(timestamp: float) -> str:
|
|
return time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(timestamp))
|