111 lines
3.3 KiB
Python
111 lines
3.3 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Fake UI per BastionGuard Secure Connection.
|
|
|
|
Simula la UI reale sul protocollo wire (Unix socket + JSON length-prefixed):
|
|
- accetta una connessione dal daemon sul socket default /tmp/bsd-daemon.sock
|
|
- logga tutti i messaggi ricevuti
|
|
- risponde automaticamente ai messaggi ask_rule con "allow once"
|
|
|
|
Uso: python3 fake_ui.py [SOCKET_PATH]
|
|
|
|
Utile per:
|
|
- testare il daemon senza compilare la UI gtkmm
|
|
- debuggare modifiche al protocollo
|
|
- smoke test di CI
|
|
"""
|
|
import json
|
|
import os
|
|
import socket
|
|
import struct
|
|
import sys
|
|
|
|
|
|
def read_exact(conn: socket.socket, n: int) -> bytes | None:
|
|
buf = b""
|
|
while len(buf) < n:
|
|
chunk = conn.recv(n - len(buf))
|
|
if not chunk:
|
|
return None
|
|
buf += chunk
|
|
return buf
|
|
|
|
|
|
def send_frame(conn: socket.socket, msg: dict) -> None:
|
|
data = json.dumps(msg).encode("utf-8")
|
|
conn.sendall(struct.pack("<I", len(data)) + data)
|
|
|
|
|
|
def handle_client(conn: socket.socket) -> None:
|
|
while True:
|
|
hdr = read_exact(conn, 4)
|
|
if not hdr:
|
|
return
|
|
length = struct.unpack("<I", hdr)[0]
|
|
if length > 4 * 1024 * 1024:
|
|
print(f"[fake-ui] frame troppo grande ({length}), chiudo", flush=True)
|
|
return
|
|
payload = read_exact(conn, length)
|
|
if not payload:
|
|
return
|
|
try:
|
|
msg = json.loads(payload.decode("utf-8"))
|
|
except Exception as e:
|
|
print(f"[fake-ui] JSON invalido: {e}", flush=True)
|
|
continue
|
|
|
|
t = msg.get("type", "?")
|
|
mid = msg.get("id", 0)
|
|
print(f"[fake-ui] recv type={t} id={mid}", flush=True)
|
|
|
|
if t == "ask_rule":
|
|
conn_info = msg.get("connection", {})
|
|
print(f"[fake-ui] app={conn_info.get('process_path')} "
|
|
f"dst={conn_info.get('dst_host') or conn_info.get('dst_ip')}"
|
|
f":{conn_info.get('dst_port')}", flush=True)
|
|
send_frame(conn, {
|
|
"type": "ask_rule_reply",
|
|
"id": mid,
|
|
"action": "allow",
|
|
"duration": "once",
|
|
"rule_name": "fake-ui-auto-allow",
|
|
})
|
|
print(f"[fake-ui] sent ask_rule_reply id={mid} action=allow", flush=True)
|
|
elif t == "alert":
|
|
text = msg.get("text", "")
|
|
if text:
|
|
print(f"[fake-ui] alert: {text}", flush=True)
|
|
|
|
|
|
def main() -> int:
|
|
sock_path = sys.argv[1] if len(sys.argv) > 1 else "/tmp/bsd-daemon.sock"
|
|
if os.path.exists(sock_path):
|
|
os.remove(sock_path)
|
|
|
|
srv = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
|
|
srv.bind(sock_path)
|
|
os.chmod(sock_path, 0o666)
|
|
srv.listen(1)
|
|
print(f"[fake-ui] in ascolto su {sock_path} (Ctrl-C per uscire)", flush=True)
|
|
|
|
try:
|
|
while True:
|
|
conn, _ = srv.accept()
|
|
print("[fake-ui] daemon connesso", flush=True)
|
|
try:
|
|
handle_client(conn)
|
|
finally:
|
|
conn.close()
|
|
print("[fake-ui] daemon disconnesso, aspetto nuova connessione", flush=True)
|
|
except KeyboardInterrupt:
|
|
print("\n[fake-ui] stop", flush=True)
|
|
finally:
|
|
try:
|
|
os.remove(sock_path)
|
|
except FileNotFoundError:
|
|
pass
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|