// Package ipc implements the local IPC server. // The UI (or any local tool) connects to 127.0.0.1:17337 and speaks // newline-delimited JSON: send IpcMessage commands, receive IpcMessage events. // // Backward compat: commands without network_id route to the first joined network. package ipc import ( "bufio" "context" "crypto/rand" "encoding/hex" "encoding/json" "fmt" "log" "net" "net/http" "strings" "time" "nhooyr.io/websocket" "github.com/waste-go/internal/crypto" "github.com/waste-go/internal/invite" "github.com/waste-go/internal/netmgr" "github.com/waste-go/internal/proto" "github.com/waste-go/internal/shares" ) // RunWS starts a WebSocket IPC server on 127.0.0.1:wsPort. // Each WebSocket connection gets the same handleClient treatment as TCP. // The OriginPatterns option allows connections from local dev servers. func RunWS(mgr *netmgr.Manager, wsPort int) error { addr := fmt.Sprintf("127.0.0.1:%d", wsPort) log.Printf("ipc: WS listening on %s", addr) return http.ListenAndServe(addr, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { conn, err := websocket.Accept(w, r, &websocket.AcceptOptions{ OriginPatterns: []string{"*"}, }) if err != nil { log.Printf("ipc: ws accept: %v", err) return } log.Printf("ipc: WS client connected") nc := websocket.NetConn(r.Context(), conn, websocket.MessageText) handleClient(nc, mgr) })) } // Run starts the IPC listener. Blocks until the listener fails. func Run(mgr *netmgr.Manager, port int) error { addr := fmt.Sprintf("127.0.0.1:%d", port) ln, err := net.Listen("tcp", addr) if err != nil { return fmt.Errorf("ipc listen on %s: %w", addr, err) } log.Printf("ipc: listening on %s", addr) for { conn, err := ln.Accept() if err != nil { return fmt.Errorf("ipc accept: %w", err) } log.Printf("ipc: UI client connected from %s", conn.RemoteAddr()) go handleClient(conn, mgr) } } func handleClient(conn net.Conn, mgr *netmgr.Manager) { defer conn.Close() events := mgr.Subscribe() defer mgr.Unsubscribe(events) writeCh := make(chan []byte, 128) done := make(chan struct{}) writerDone := make(chan struct{}) // Writer goroutine. go func() { defer close(writerDone) w := bufio.NewWriter(conn) for line := range writeCh { line = append(line, '\n') if _, err := w.Write(line); err != nil { return } w.Flush() } }() // Event pusher — forwards Manager events to the UI client. go func() { defer func() { recover() }() //nolint:errcheck for { select { case evt, ok := <-events: if !ok { return } line, err := json.Marshal(evt) if err != nil { continue } select { case writeCh <- line: case <-done: return } case <-done: return } } }() send := func(msg proto.IpcMessage) { defer func() { recover() }() //nolint:errcheck line, err := json.Marshal(msg) if err != nil { return } select { case writeCh <- line: case <-done: } } // Send initial state snapshot. send(stateSnapshot(mgr)) scanner := bufio.NewScanner(conn) for scanner.Scan() { var cmd proto.IpcMessage if err := json.Unmarshal(scanner.Bytes(), &cmd); err != nil { log.Printf("ipc: bad command: %v", err) continue } switch cmd.Type { case proto.CmdJoinNetwork: var ( netID string err error ) switch { case cmd.NetworkName != "": netID, err = mgr.Join(cmd.NetworkName, cmd.ShareDir) case len(cmd.NetworkHash) == 64: netID, err = mgr.JoinByHash(cmd.NetworkHash, cmd.ShareDir) default: send(errMsg("join_network: network_name or network_hash (64 hex chars) required")) continue } if err != nil { send(errMsg(fmt.Sprintf("join_network: %v", err))) continue } if n, ok := mgr.Get(netID); ok { if cmd.RequireInvite { n.Mesh.RequireInvite = true } if cmd.InviteString != "" { n.Mesh.InviteString = cmd.InviteString } } _ = netID case proto.CmdLeaveNetwork: if cmd.NetworkID != "" { mgr.Leave(cmd.NetworkID) } else { // Backward compat: leave the first joined network. if n := mgr.Default(); n != nil { mgr.Leave(n.ID) } } case proto.CmdSendMessage: n := mgr.Resolve(cmd.NetworkID) if n == nil { send(errMsg("send_message: not joined to any network")) continue } ts := time.Now().UnixMilli() if cmd.To != nil { // DM → spec "pm" type: flat {type, mid, text, ts} on the wire mid := randomHex(16) wire, err := json.Marshal(proto.PeerMessage{ Type: proto.MsgPm, Mid: mid, Text: cmd.Body, Ts: ts, }) if err != nil { continue } n.Mesh.SendTo(*cmd.To, wire) // Store locally with dm: room convention local := &proto.ChatMessage{ Mid: mid, From: n.Identity.PeerID(), To: cmd.To, Room: "dm:" + (*cmd.To).Short(), Text: cmd.Body, Ts: ts, } n.Mesh.SaveMessage(local) n.Mesh.Emit(proto.IpcMessage{ Type: proto.EvtMessageReceived, NetworkID: n.ID, Message: local, }) } else { // Group chat → spec "chat" type: flat {type, mid, room, text, ts} mid := randomHex(16) msgID := proto.ComputeMsgID(n.Identity.PeerID(), cmd.Room, ts, cmd.Body) wire, err := json.Marshal(proto.PeerMessage{ Type: proto.MsgChat, Mid: mid, Room: cmd.Room, Text: cmd.Body, Ts: ts, }) if err != nil { continue } n.Mesh.Broadcast(wire) local := &proto.ChatMessage{ Mid: mid, MsgID: msgID, From: n.Identity.PeerID(), Room: cmd.Room, Text: cmd.Body, Ts: ts, } n.Mesh.SaveMessage(local) n.Mesh.Emit(proto.IpcMessage{ Type: proto.EvtMessageReceived, NetworkID: n.ID, Message: local, }) } case proto.CmdCreateRoom: n := mgr.Resolve(cmd.NetworkID) if n == nil { send(errMsg("create_room: not joined to any network")) continue } name := strings.TrimSpace(cmd.Room) if name == "" || name == "general" { send(errMsg("create_room: room name is required and cannot be 'general'")) continue } if err := n.Store.SaveRoom(name); err != nil { send(errMsg(fmt.Sprintf("create_room: %v", err))) continue } send(proto.IpcMessage{Type: proto.EvtRoomCreated, NetworkID: n.ID, Room: name}) case proto.CmdGetState: send(stateSnapshot(mgr)) case proto.CmdGetFileList: n := mgr.Resolve(cmd.NetworkID) if n == nil { send(errMsg("get_file_list: not joined to any network")) continue } if cmd.PeerID == nil || *cmd.PeerID == n.Identity.PeerID() { send(proto.IpcMessage{ Type: proto.EvtFileList, NetworkID: n.ID, PeerID: ptr(n.Identity.PeerID()), Files: mgr.ScanAllShares(n.ID), }) } else { req, err := json.Marshal(proto.PeerMessage{Type: proto.MsgFileListReq}) if err != nil { continue } if !n.Mesh.SendTo(*cmd.PeerID, req) { send(errMsg(fmt.Sprintf("get_file_list: peer %s not connected", (*cmd.PeerID).Short()))) } } case proto.CmdAddShare: if cmd.Path == "" { send(errMsg("add_share: path is required")) continue } networks := cmd.ShareNetworks if len(networks) == 0 { networks = []string{"*"} } if err := mgr.Shares.Add(shares.Share{Path: cmd.Path, Networks: networks}); err != nil { send(errMsg(fmt.Sprintf("add_share: %v", err))) continue } send(sharesListMsg(mgr)) case proto.CmdRemoveShare: if cmd.Path == "" { send(errMsg("remove_share: path is required")) continue } if err := mgr.Shares.Remove(cmd.Path); err != nil { send(errMsg(fmt.Sprintf("remove_share: %v", err))) continue } send(sharesListMsg(mgr)) case proto.CmdListShares: send(sharesListMsg(mgr)) case proto.CmdGenerateInvite: n := mgr.Resolve(cmd.NetworkID) if n == nil { send(errMsg("generate_invite: not currently joined to a network")) continue } if mgr.AnchorURL() == "" { send(errMsg("generate_invite: daemon was started without -anchor flag")) continue } inv, err := invite.EncodeSigned(mgr.AnchorURL(), n.Name, n.Identity) if err != nil { send(errMsg(fmt.Sprintf("generate_invite: %v", err))) continue } send(proto.IpcMessage{ Type: proto.EvtInviteGenerated, NetworkID: n.ID, InviteGenerated: inv, }) case proto.CmdSetShareDir: n := mgr.Resolve(cmd.NetworkID) if n == nil { send(errMsg("set_share_dir: not joined to any network")) continue } if !mgr.SetShareDir(n.ID, cmd.Path) { send(errMsg("set_share_dir: network not found")) continue } case proto.CmdSetDownloadDir: n := mgr.Resolve(cmd.NetworkID) if n == nil { send(errMsg("set_download_dir: not joined to any network")) continue } if cmd.Path == "" { send(errMsg("set_download_dir: path is required")) continue } if !mgr.SetDownloadDir(n.ID, cmd.Path) { send(errMsg("set_download_dir: network not found")) continue } send(stateSnapshot(mgr)) case proto.CmdSendFile: n := mgr.Resolve(cmd.NetworkID) if n == nil { send(errMsg("send_file: not joined to any network")) continue } if cmd.PeerID == nil { send(errMsg("send_file: peer_id is required")) continue } if cmd.Path == "" { send(errMsg("send_file: path is required")) continue } if err := n.Mesh.OfferFile(*cmd.PeerID, cmd.Path); err != nil { send(errMsg(fmt.Sprintf("send_file: %v", err))) } case proto.CmdExportIdentity: if cmd.Passphrase == "" { send(errMsg("export_identity: passphrase is required")) continue } blob, err := crypto.ExportIdentity(mgr.MasterIdentity(), cmd.Passphrase) if err != nil { send(errMsg(fmt.Sprintf("export_identity: %v", err))) continue } send(proto.IpcMessage{ Type: proto.EvtIdentityExported, Backup: string(blob), }) case proto.CmdImportIdentity: if cmd.Passphrase == "" || cmd.Backup == "" { send(errMsg("import_identity: passphrase and backup are required")) continue } _, err := crypto.ImportIdentity([]byte(cmd.Backup), cmd.Passphrase) if err != nil { send(errMsg(fmt.Sprintf("import_identity: %v", err))) continue } // Import is intentionally read-only here: returns the decrypted identity // for the caller to verify before committing. Actual on-disk replacement // requires a daemon restart with --import flag (see cmd/daemon). send(proto.IpcMessage{Type: proto.EvtIdentityImported}) default: send(errMsg(fmt.Sprintf("unknown command: %s", cmd.Type))) } } close(done) close(writeCh) <-writerDone // wait for writer to flush before conn.Close() fires log.Printf("ipc: UI client disconnected") } // stateSnapshot builds a state_snapshot covering all joined networks. // Backward compat: local_peer and connected_peers are populated from the first network. func stateSnapshot(mgr *netmgr.Manager) proto.IpcMessage { all := mgr.All() master := mgr.MasterIdentity() msg := proto.IpcMessage{ Type: proto.EvtStateSnapshot, Rooms: []string{"general"}, MasterAlias: master.Alias, MasterID: string(master.PeerID()), } var netInfos []proto.NetworkInfo for _, n := range all { pi := n.Identity.PeerInfo() netInfos = append(netInfos, proto.NetworkInfo{ NetworkID: n.ID, NetworkName: n.Name, LocalPeer: &pi, ShareDir: n.Mesh.ShareDir, DownloadDir: n.Mesh.DownloadDir, }) } msg.Networks = netInfos // Backward-compat fields — populated from the first network when one exists. if len(all) > 0 { pi := all[0].Identity.PeerInfo() msg.LocalPeer = &pi msg.ConnectedPeers = all[0].Mesh.ConnectedPeers() if extra, err := all[0].Store.Rooms(); err == nil { for _, r := range extra { msg.Rooms = append(msg.Rooms, r) } } } return msg } func errMsg(s string) proto.IpcMessage { return proto.IpcMessage{Type: proto.EvtError, ErrorMessage: s} } func sharesListMsg(mgr *netmgr.Manager) proto.IpcMessage { all := mgr.Shares.All() entries := make([]proto.ShareEntry, len(all)) for i, sh := range all { entries[i] = proto.ShareEntry{Path: sh.Path, Networks: sh.Networks} } return proto.IpcMessage{Type: proto.EvtSharesList, Shares: entries} } // ensure shares import is used var _ = shares.Share{} func ptr[T any](v T) *T { return &v } func randomHex(n int) string { b := make([]byte, n) rand.Read(b) //nolint:errcheck return hex.EncodeToString(b) } // autoJoin is called by the daemon when --join is provided at startup. // It joins the network before the IPC listener starts accepting clients. func AutoJoin(ctx context.Context, mgr *netmgr.Manager, networkName string) { _ = ctx // Manager owns the context internally if _, err := mgr.Join(networkName, ""); err != nil { log.Printf("ipc: auto-join %q failed: %v", networkName, err) } }