fake_ui.py

Go to the documentation of this file.

#!/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())