import RNS import RNS.vendor.umsgpack as msgpack import collections import os import shutil import time import nomadnet import LXMF import urwid from datetime import datetime, timedelta from nomadnet.Directory import DirectoryEntry from LXMF import pn_announce_data_is_valid, PN_META_NAME from nomadnet.Conversation import ConversationMessage from nomadnet.util import strip_modifiers from nomadnet.util import sanitize_name from RNS.Utilities.rngit.util import MarkdownToMicron from RNS.Utilities.rngit.highlight import SyntaxHighlighter from .MicronParser import markup_to_attrmaps from .Helpers import ClickableIcon, osc52_copy from nomadnet.util import strip_modifiers, strip_micron, strip_escaped_micron, unescape_micron, strip_non_formatting_tags from nomadnet.ui import THEME_DARK, THEME_LIGHT def relative_time(timestamp): now = time.time() delta = now - timestamp if delta < 0: return "just now" elif delta < 60: return "just now" elif delta < 3600: m = int(delta / 60) return str(m)+"m ago" elif delta < 86400: h = int(delta / 3600) return str(h)+"h ago" elif delta < 172800: return "yesterday" elif delta < 604800: d = int(delta / 86400) return str(d)+"d ago" elif delta < 2592000: w = int(delta / 604800) return str(w)+"w ago" else: return datetime.fromtimestamp(timestamp).strftime("%Y-%m-%d") def _format_size(size): if size < 1024: return str(size)+" B" elif size < 1048576: return str(round(size/1024, 1))+" KB" else: return str(round(size/1048576, 1))+" MB" from nomadnet.vendor.additional_urwid_widgets import IndicativeListBox class ConversationListDisplayShortcuts(): def __init__(self, app): self.app = app self.widget = urwid.AttrMap(urwid.Text("[C-e] Peer Info [C-x] Delete [C-r] Sync [C-n] New [C-u] Ingest URI [C-o] Sort [C-p] My LXMF [C-g] Fullscreen"), "shortcutbar") class ConversationDisplayShortcuts(): def __init__(self, app): self.app = app self.widget = urwid.AttrMap(urwid.Text("[C-d] Send [C-p] Paper Msg [C-t] Title [C-a] Attach [C-s] Save [C-k] Clear [C-w] Close [C-u] Purge [C-x] Clear History [C-o] Sort"), "shortcutbar") class TabButton(urwid.Button): button_left = urwid.Text("[") button_right = urwid.Text("]") class ConversationsArea(urwid.LineBox): def keypress(self, size, key): if key == "ctrl e": self.delegate.edit_selected_in_directory() elif key == "ctrl x": self.delegate.delete_selected_conversation() elif key == "ctrl n": self.delegate.new_conversation() elif key == "ctrl u": self.delegate.ingest_lxm_uri() elif key == "ctrl r": self.delegate.sync_conversations() elif key == "ctrl g": self.delegate.toggle_fullscreen() elif key == "ctrl o": self.delegate.toggle_list_sort() elif key == "ctrl p": self.delegate.show_my_qr() elif key == "tab": self.delegate.app.ui.main_display.frame.focus_position = "header" elif key == "up": if self.delegate.ilb.body_is_empty(): self.delegate.app.ui.main_display.frame.focus_position = "header" return None result = super(ConversationsArea, self).keypress(size, key) if result == "up": self.delegate.app.ui.main_display.frame.focus_position = "header" return None return result else: return super(ConversationsArea, self).keypress(size, key) class PropNodePicker(urwid.WidgetWrap): def __init__(self, options, current_hash, on_change): self._options = list(options) self._current = current_hash self._on_change = on_change self._pile = urwid.Pile([]) super().__init__(self._pile) self._show_collapsed() def _label_for(self, h): for ph, label in self._options: if ph == h: return label if h is not None: return "<"+RNS.hexrep(h, delimit=False)+">" return "(select propagation node)" def _show_collapsed(self): btn = urwid.Button("▾ "+self._label_for(self._current)) urwid.connect_signal(btn, "click", self._on_expand_click) self._pile.contents = [ (urwid.AttrMap(btn, None, focus_map="list_focus"), self._pile.options()), ] def _on_expand_click(self, _btn): self._show_expanded() def _on_back_click(self, _btn): self._show_collapsed() def _make_row_click(self, picked_hash): def _click(_btn): try: self._current = picked_hash self._on_change(picked_hash) except Exception as e: RNS.log("Propagation node change handler failed: "+str(e), RNS.LOG_ERROR) self._show_collapsed() return _click def _show_expanded(self): rows = [] for ph, label in self._options: marker = "● " if ph == self._current else "○ " row = urwid.Button(marker+label) urwid.connect_signal(row, "click", self._make_row_click(ph)) rows.append(urwid.AttrMap(row, None, focus_map="list_focus")) if not rows: rows.append(urwid.Text(" (no propagation nodes seen yet)")) list_height = min(10, max(3, len(rows))) listbox = urwid.ListBox(urwid.SimpleFocusListWalker(rows)) boxed = urwid.BoxAdapter(listbox, list_height) back = urwid.Button("◀ Back") urwid.connect_signal(back, "click", self._on_back_click) self._pile.contents = [ (urwid.AttrMap(back, None, focus_map="list_focus"), self._pile.options()), (boxed, self._pile.options()), ] class DialogLineBox(urwid.LineBox): def keypress(self, size, key): if key == "esc": if hasattr(self.delegate, "update_conversation_list"): self.delegate.update_conversation_list() elif hasattr(self.delegate, "dialog_active"): self.delegate.dialog_active = False self.delegate.conversation_changed(None) else: return super(DialogLineBox, self).keypress(size, key) class ConversationsDisplay(): list_width = 0.33 given_list_width = 52 cached_conversation_widgets = {} SORT_RECENT = 0 SORT_NAME = 1 LIST_FILTER_TRUSTED = "trusted" LIST_FILTER_UNTRUSTED = "untrusted" def __init__(self, app): self.app = app self.dialog_open = False self.sync_dialog = None self.currently_displayed_conversation = None self.list_sort_mode = ConversationsDisplay.SORT_RECENT self.list_filter = ConversationsDisplay.LIST_FILTER_TRUSTED self.show_blocked = False def disp_list_shortcuts(sender, arg1, arg2): self.shortcuts_display = self.list_shortcuts self.app.ui.main_display.update_active_shortcuts() self._build_persistent_listbox() self.update_listbox() self.columns_widget = urwid.Columns( [ # (urwid.WEIGHT, ConversationsDisplay.list_width, self.listbox), # (urwid.WEIGHT, 1-ConversationsDisplay.list_width, self.make_conversation_widget(None)) (ConversationsDisplay.given_list_width, self.listbox), (urwid.WEIGHT, 1, self.make_conversation_widget(None)) ], dividechars=0, focus_column=0, box_columns=[0] ) self.list_shortcuts = ConversationListDisplayShortcuts(self.app) self.editor_shortcuts = ConversationDisplayShortcuts(self.app) self.shortcuts_display = self.list_shortcuts self.widget = urwid.WidgetPlaceholder(self.columns_widget) self._pending_actions = collections.deque() self._wake_fd = None try: self._wake_fd = self.app.ui.loop.watch_pipe(self._process_pending) except Exception: pass nomadnet.Conversation.created_callback = lambda: self._wake(self.update_conversation_list) try: self.app.ui.loop.set_alarm_in(30.0, self._refresh_sync_status) except Exception: pass def _process_pending(self, data): while True: try: action = self._pending_actions.popleft() except IndexError: break try: action() except Exception as e: RNS.log("Conversations UI action failed: "+str(e), RNS.LOG_ERROR) return True def _wake(self, action): self._pending_actions.append(action) if self._wake_fd is not None: try: os.write(self._wake_fd, b".") return except Exception: pass try: self.app.ui.loop.set_alarm_in(0.0, lambda l, d: self._process_pending(None)) except Exception: pass def focus_change_event(self): if not self.dialog_open: self.update_conversation_list() def toggle_list_sort(self): if self.list_sort_mode == ConversationsDisplay.SORT_RECENT: self.list_sort_mode = ConversationsDisplay.SORT_NAME else: self.list_sort_mode = ConversationsDisplay.SORT_RECENT self.update_conversation_list() def _conversation_filter_predicate(self, conversation): try: trust_level = conversation[2] except Exception: return False if self.list_filter == ConversationsDisplay.LIST_FILTER_UNTRUSTED: return trust_level in (DirectoryEntry.UNTRUSTED, DirectoryEntry.WARNING, DirectoryEntry.UNKNOWN) return trust_level == DirectoryEntry.TRUSTED def _set_filter(self, key): if self.list_filter == key: return self.list_filter = key try: self.update_conversation_list() except Exception as e: RNS.log("Failed to apply conversation filter: "+str(e), RNS.LOG_ERROR) def _on_show_blocked_change(self, _cb, new_state): self.show_blocked = new_state try: self.update_conversation_list() except Exception as e: RNS.log("Failed to toggle show-blocked: "+str(e), RNS.LOG_ERROR) def _apply_pile_layout(self): pack_opts = self.pile.options('pack') weight_opts = self.pile.options('weight', 1) items = [(self.tab_bar, pack_opts)] if self.list_filter == ConversationsDisplay.LIST_FILTER_UNTRUSTED: items.append((self.show_blocked_checkbox, pack_opts)) items.append((self.ilb, weight_opts)) items.append((self.sync_status_text, pack_opts)) try: prev_focus = self.pile.focus_position except Exception: prev_focus = None self.pile.contents = items try: if prev_focus is not None and prev_focus < len(items): self.pile.focus_position = prev_focus except Exception: pass def _blocked_row_widget(self, dest_hash): g = self.app.ui.glyphs entry = self.app.directory.find(dest_hash) display_name = None if entry is not None and getattr(entry, "display_name", None): display_name = strip_modifiers(entry.display_name) if not display_name: display_name = RNS.prettyhexrep(dest_hash) label = " "+g["cross"]+" [blocked] "+display_name+" <"+RNS.hexrep(dest_hash, delimit=False)+">" widget = ListEntry(label) urwid.connect_signal(widget, "click", self._unblock_dialog, dest_hash) attr = urwid.AttrMap(widget, "list_untrusted", "list_focus_untrusted") attr.blocked_dest_hash = dest_hash return attr def _unblock_dialog(self, _sender, dest_hash): self.dialog_open = True def dismiss(_b): self.dialog_open = False self.update_conversation_list() def confirmed(_b): self.dialog_open = False try: self.app.unblock_destination(dest_hash) except Exception as e: RNS.log("Unblock failed: "+str(e), RNS.LOG_ERROR) self.update_conversation_list() try: who = self.app.directory.simplest_display_str(dest_hash) except Exception: who = RNS.hexrep(dest_hash, delimit=False) dialog = DialogLineBox( urwid.Pile([ urwid.Text(""), urwid.Text("Unblock "+str(who)+"?\n\nThis lifts the RNS blackhole on the peer's identity\nand removes them from your ignored list.\n", align=urwid.CENTER), urwid.Columns([ (urwid.WEIGHT, 0.45, urwid.Button("Yes, unblock", on_press=confirmed)), (urwid.WEIGHT, 0.10, urwid.Text("")), (urwid.WEIGHT, 0.45, urwid.Button("Cancel", on_press=dismiss)), ]), ]), title="Confirm unblock" ) dialog.delegate = self bottom = self.listbox overlay = urwid.Overlay(dialog, bottom, align=urwid.CENTER, width=urwid.RELATIVE_100, valign=urwid.MIDDLE, height=urwid.PACK, left=2, right=2) try: self.columns_widget.contents[0] = ( overlay, self.columns_widget.options(urwid.GIVEN, ConversationsDisplay.given_list_width), ) self.columns_widget.focus_position = 0 except Exception: pass def _build_persistent_listbox(self): self.tab_trusted = TabButton("Trusted (0)", on_press=lambda _b: self._set_filter(ConversationsDisplay.LIST_FILTER_TRUSTED)) self.tab_untrusted = TabButton("Untrusted (0)", on_press=lambda _b: self._set_filter(ConversationsDisplay.LIST_FILTER_UNTRUSTED)) self.tab_bar = urwid.Columns([ ('weight', 1, self.tab_trusted), ('weight', 1, self.tab_untrusted), ], dividechars=1) self.show_blocked_checkbox = urwid.CheckBox("Show blocked (0)", state=self.show_blocked) urwid.connect_signal(self.show_blocked_checkbox, "change", self._on_show_blocked_change) self.ilb = IndicativeListBox( [urwid.Text("")], on_selection_change=self.conversation_list_selection, initialization_is_selection_change=False, highlight_offFocus="list_off_focus", ) self.sync_status_text = urwid.AttrMap(urwid.Text(self._sync_status_line(), align=urwid.LEFT), "shortcutbar") self.pile = urwid.Pile([ ('pack', self.tab_bar), ('weight', 1, self.ilb), ('pack', self.sync_status_text), ]) try: self.pile.focus_position = 1 except Exception: pass self.listbox = ConversationsArea(self.pile, title="Conversations") self.listbox.delegate = self def update_listbox(self): if not hasattr(self, "pile"): self._build_persistent_listbox() try: conversations = self.app.conversations() except Exception as e: RNS.log("Failed to enumerate conversations: "+str(e), RNS.LOG_ERROR) conversations = [] try: if self.list_sort_mode == ConversationsDisplay.SORT_NAME: conversations.sort(key=lambda e: (e[3].lower(), e[0])) except Exception: pass def _is_pinned(c): try: entry = self.app.directory.find(bytes.fromhex(c[0])) return entry is not None and entry.sort_rank is not None except Exception: return False try: conversations = sorted(conversations, key=lambda c: 0 if _is_pinned(c) else 1) except Exception: pass glyphs = self.app.ui.glyphs def _alerts(c): return bool(c[4]) or (len(c) > 6 and bool(c[6])) trusted_count = sum(1 for c in conversations if c[2] == DirectoryEntry.TRUSTED) untrusted_count = sum(1 for c in conversations if c[2] in (DirectoryEntry.UNTRUSTED, DirectoryEntry.WARNING, DirectoryEntry.UNKNOWN)) trusted_unread = sum(1 for c in conversations if c[2] == DirectoryEntry.TRUSTED and _alerts(c)) untrusted_unread = sum(1 for c in conversations if c[2] in (DirectoryEntry.UNTRUSTED, DirectoryEntry.WARNING, DirectoryEntry.UNKNOWN) and _alerts(c)) def _label(name, total, unread): if unread: return f"{name} ({total}) {glyphs['unread']} {unread}" return f"{name} ({total})" self.tab_trusted.set_label(_label("Trusted", trusted_count, trusted_unread)) self.tab_untrusted.set_label(_label("Untrusted", untrusted_count, untrusted_unread)) filtered = [c for c in conversations if self._conversation_filter_predicate(c)] conversation_list_widgets = [] for conversation in filtered: try: conversation_list_widgets.append(self.conversation_list_widget(conversation)) except Exception as e: try: hh = conversation[0] except Exception: hh = "?" RNS.log("Skipping conversation row for "+str(hh)+": "+str(e), RNS.LOG_ERROR) blocked_count = len(self.app.ignored_list) if hasattr(self.app, "ignored_list") else 0 try: self.show_blocked_checkbox.set_label(f"Show blocked ({blocked_count})") except Exception: pass if self.list_filter == ConversationsDisplay.LIST_FILTER_UNTRUSTED and self.show_blocked: for blocked_hash in list(self.app.ignored_list): try: conversation_list_widgets.append(self._blocked_row_widget(blocked_hash)) except Exception as e: RNS.log("Skipping blocked row: "+str(e), RNS.LOG_ERROR) empty_placeholder = False if not conversation_list_widgets: empty_label = { ConversationsDisplay.LIST_FILTER_TRUSTED: "No trusted conversations", ConversationsDisplay.LIST_FILTER_UNTRUSTED: "No untrusted conversations", }.get(self.list_filter, "No conversations") conversation_list_widgets = [urwid.Text(empty_label, align='center')] empty_placeholder = True self.list_widgets = conversation_list_widgets try: self.ilb.set_body(conversation_list_widgets) except Exception as e: RNS.log("Failed to populate conversation list: "+str(e), RNS.LOG_ERROR) self._apply_pile_layout() if empty_placeholder: try: self.pile.focus_position = 0 except Exception: pass try: self.sync_status_text.original_widget.set_text(self._sync_status_line()) except Exception: pass def _sync_status_line(self): try: last = self.app.peer_settings.get("last_lxmf_sync") except Exception: last = None if not last: when = "never" else: try: when = relative_time(float(last)) except Exception: when = "unknown" node_label = None try: pn_hash = self.app.get_default_propagation_node() if pn_hash is not None: pn_ident = RNS.Identity.recall(pn_hash) if pn_ident is not None: node_dest = RNS.Destination.hash_from_name_and_identity("nomadnetwork.node", pn_ident) entry = self.app.directory.find(node_dest) if entry is not None and getattr(entry, "display_name", None): node_label = strip_modifiers(str(entry.display_name)) or None if node_label is None: node_label = "<"+RNS.hexrep(pn_hash, delimit=False)[:8]+"…>" except Exception: node_label = None line = " Last sync: "+when if node_label: line += " ("+node_label+")" return line def _refresh_sync_status(self, _loop=None, _data=None): try: if hasattr(self, "sync_status_text") and self.sync_status_text is not None: self.sync_status_text.original_widget.set_text(self._sync_status_line()) except Exception: pass try: self.app.ui.loop.set_alarm_in(30.0, self._refresh_sync_status) except Exception: pass def delete_selected_conversation(self): self.dialog_open = True item = self.ilb.get_selected_item() if item == None: return source_hash = item.source_hash def dismiss_dialog(sender): self.update_conversation_list() self.dialog_open = False def confirmed(sender): self.dialog_open = False self.delete_conversation(source_hash) nomadnet.Conversation.delete_conversation(source_hash, self.app) self.update_conversation_list() dialog = DialogLineBox( urwid.Pile([ urwid.Text( "Delete conversation with\n"+self.app.directory.simplest_display_str(bytes.fromhex(source_hash))+"\n", align=urwid.CENTER, ), urwid.Columns([ (urwid.WEIGHT, 0.45, urwid.Button("Yes", on_press=confirmed)), (urwid.WEIGHT, 0.1, urwid.Text("")), (urwid.WEIGHT, 0.45, urwid.Button("No", on_press=dismiss_dialog)), ]) ]), title="?" ) dialog.delegate = self bottom = self.listbox overlay = urwid.Overlay( dialog, bottom, align=urwid.CENTER, width=urwid.RELATIVE_100, valign=urwid.MIDDLE, height=urwid.PACK, left=2, right=2, ) # options = self.columns_widget.options(urwid.WEIGHT, ConversationsDisplay.list_width) options = self.columns_widget.options(urwid.GIVEN, ConversationsDisplay.given_list_width) self.columns_widget.contents[0] = (overlay, options) def _refresh_open_conversation_widget(self, source_hash_text): widget = ConversationsDisplay.cached_conversation_widgets.get(source_hash_text) if widget is None: return try: widget._trust_banner_dismissed = False except Exception: pass try: widget._refresh_trust_banner() except Exception: pass try: widget._update_peer_info() except Exception: pass try: widget.update_message_widgets(replace=True) except Exception: pass def show_my_qr(self): try: addr = RNS.hexrep(self.app.lxmf_destination.hash, delimit=False) except Exception: return try: display = self.app.peer_settings.get("display_name") or "My LXMF" except Exception: display = "My LXMF" self.show_qr_dialog(addr, title=display) def show_qr_dialog(self, data, title=None): qr_text = None try: import qrcode try: qr = qrcode.QRCode(error_correction=qrcode.constants.ERROR_CORRECT_L, box_size=1, border=1) qr.add_data(data) qr.make() import io buf = io.StringIO() qr.print_ascii(out=buf, invert=False) qr_text = buf.getvalue().rstrip("\n") except Exception as e: RNS.log("QR generation failed: "+str(e), RNS.LOG_ERROR) qr_text = None except Exception: qr_text = None def dismiss(_b): self._restore_listbox_pane() rows = [urwid.Text("")] if qr_text is not None: rows.append(urwid.Text(qr_text, align=urwid.CENTER)) rows.append(urwid.Text("")) else: rows.append(urwid.Text("LXMF destination address:", align=urwid.CENTER)) rows.append(urwid.Text("")) rows += [ urwid.Text("< "+data+" >", align=urwid.CENTER), urwid.Text(""), urwid.Columns([ (urwid.WEIGHT, 1, urwid.Text("")), (12, urwid.Button("Close", on_press=dismiss)), (urwid.WEIGHT, 1, urwid.Text("")), ]), urwid.Text(""), ] dialog_title = "LXMF Address" if qr_text is None else "QR Code" dialog = DialogLineBox(urwid.Pile(rows), title=dialog_title) dialog.delegate = self self._overlay_dialog(dialog) def _overlay_dialog(self, dialog): overlay = urwid.Overlay( dialog, self.columns_widget, align=urwid.CENTER, width=(urwid.RELATIVE, 70), valign=urwid.MIDDLE, height=urwid.PACK, min_width=44, ) try: self.widget.original_widget = overlay self.dialog_open = True except Exception: pass def _restore_listbox_pane(self): try: self.widget.original_widget = self.columns_widget self.dialog_open = False self.update_conversation_list() except Exception: pass def _ping_peer_from_dialog(self, source_hash_text, status_widget, ping_button): try: dest = bytes.fromhex(source_hash_text) except Exception: status_widget.set_text(("error_text", "Invalid address")) return identity = RNS.Identity.recall(dest) if identity is None: status_widget.set_text(("error_text", "Identity unknown; query first")) return if not RNS.Transport.has_path(dest): status_widget.set_text("No path; requesting…") try: RNS.Transport.request_path(dest) except Exception: pass status_widget.set_text("Pinging…") ping_button.set_label("Pinging…") started_at = time.time() def schedule_ui(fn): try: self.app.ui.loop.set_alarm_in(0, lambda *_: fn()) except Exception: try: fn() except Exception: pass def on_established(link): elapsed_ms = int((time.time() - started_at) * 1000) try: hops = RNS.Transport.hops_to(dest) if hops is None or hops >= RNS.Transport.PATHFINDER_M: hops_str = "" else: hops_str = f" ({hops} hop{'s' if hops != 1 else ''})" except Exception: hops_str = "" def update(): status_widget.set_text(f"Pong in {elapsed_ms} ms{hops_str}") ping_button.set_label("Ping") schedule_ui(update) try: link.teardown() except Exception: pass def on_closed(link): try: if getattr(link, "status", None) == RNS.Link.ACTIVE: return except Exception: pass def update(): if status_widget.text.strip() in ("Pinging…", ""): status_widget.set_text(("error_text", "Ping failed (no link)")) ping_button.set_label("Ping") schedule_ui(update) try: destination = RNS.Destination(identity, RNS.Destination.OUT, RNS.Destination.SINGLE, "lxmf", "delivery") RNS.Link(destination, established_callback=on_established, closed_callback=on_closed) except Exception as e: status_widget.set_text(("error_text", f"Ping init failed: {e}")) ping_button.set_label("Ping") def _block_peer_from_dialog(self, source_hash_text): try: dest = bytes.fromhex(source_hash_text) except Exception: return def cancel_block(_b): self.update_conversation_list() self.dialog_open = False def confirm_block(_b): try: self.app.block_destination(dest, reason="user-blocked from peer info dialog") except Exception as e: RNS.log("Block failed: "+str(e), RNS.LOG_ERROR) try: self.delete_conversation(source_hash_text) nomadnet.Conversation.delete_conversation(source_hash_text, self.app) except Exception: pass self.update_conversation_list() self.dialog_open = False try: who = self.app.directory.simplest_display_str(dest) except Exception: who = source_hash_text confirm_dialog = DialogLineBox( urwid.Pile([ urwid.Text(""), urwid.Text("Block "+str(who)+"?\n\nThis blackholes the peer's identity in Reticulum,\nadds them to your ignored list, and deletes any\nconversation with them.\n", align=urwid.CENTER), urwid.Columns([ (urwid.WEIGHT, 0.45, urwid.Button("Yes, block", on_press=confirm_block)), (urwid.WEIGHT, 0.10, urwid.Text("")), (urwid.WEIGHT, 0.45, urwid.Button("Cancel", on_press=cancel_block)), ]), ]), title="Confirm block" ) confirm_dialog.delegate = self bottom = self.listbox overlay = urwid.Overlay(confirm_dialog, bottom, align=urwid.CENTER, width=urwid.RELATIVE_100, valign=urwid.MIDDLE, height=urwid.PACK, left=2, right=2) try: self.columns_widget.contents[0] = ( overlay, self.columns_widget.options(urwid.GIVEN, ConversationsDisplay.given_list_width), ) self.columns_widget.focus_position = 0 self.dialog_open = True except Exception: pass def edit_selected_in_directory(self): g = self.app.ui.glyphs self.dialog_open = True item = self.ilb.get_selected_item() if item == None: self.dialog_open = False return source_hash_text = getattr(item, "source_hash", None) if source_hash_text is None: blocked = getattr(item, "blocked_dest_hash", None) if isinstance(blocked, (bytes, bytearray)): source_hash_text = RNS.hexrep(blocked, delimit=False) if source_hash_text is None: self.dialog_open = False return display_name = getattr(item, "display_name", None) if display_name is None: try: display_name = self.app.directory.display_name(bytes.fromhex(source_hash_text)) except Exception: display_name = None if display_name is None: display_name = "" e_id = urwid.Edit(caption="Addr : ",edit_text=source_hash_text) t_id = urwid.Text("Addr : "+source_hash_text) e_name = urwid.Edit(caption="Name : ",edit_text=display_name) e_copy = urwid.Edit(caption="Copy : ", edit_text=source_hash_text) selected_id_widget = t_id untrusted_selected = False unknown_selected = True trusted_selected = False direct_selected = True propagated_selected = False pinned_initial = False notes_initial = "" try: existing_entry = self.app.directory.find(bytes.fromhex(source_hash_text)) if existing_entry: trust_level = self.app.directory.trust_level(bytes.fromhex(source_hash_text)) if trust_level == DirectoryEntry.UNTRUSTED: untrusted_selected = True unknown_selected = False trusted_selected = False elif trust_level == DirectoryEntry.UNKNOWN: untrusted_selected = False unknown_selected = True trusted_selected = False elif trust_level == DirectoryEntry.TRUSTED: untrusted_selected = False unknown_selected = False trusted_selected = True if self.app.directory.preferred_delivery(bytes.fromhex(source_hash_text)) == DirectoryEntry.PROPAGATED: direct_selected = False propagated_selected = True pinned_initial = existing_entry.sort_rank is not None notes_initial = getattr(existing_entry, "notes", "") or "" except Exception as e: pass e_notes = urwid.Edit(caption="Notes: ", edit_text=notes_initial, multiline=True) cb_pin = urwid.CheckBox("Pin to top", state=pinned_initial) trust_button_group = [] r_untrusted = urwid.RadioButton(trust_button_group, "Untrusted", state=untrusted_selected) r_unknown = urwid.RadioButton(trust_button_group, "Unknown", state=unknown_selected) r_trusted = urwid.RadioButton(trust_button_group, "Trusted", state=trusted_selected) method_button_group = [] r_direct = urwid.RadioButton(method_button_group, "Deliver directly", state=direct_selected) r_propagated = urwid.RadioButton(method_button_group, "Use propagation nodes", state=propagated_selected) def dismiss_dialog(sender): self.dialog_open = False self.update_conversation_list() def confirmed(sender): try: display_name = e_name.get_edit_text() source_hash = bytes.fromhex(e_id.get_edit_text()) trust_level = DirectoryEntry.UNTRUSTED if r_unknown.state == True: trust_level = DirectoryEntry.UNKNOWN elif r_trusted.state == True: trust_level = DirectoryEntry.TRUSTED delivery = DirectoryEntry.DIRECT if r_propagated.state == True: delivery = DirectoryEntry.PROPAGATED sort_rank = 0 if cb_pin.state else None notes_value = e_notes.get_edit_text() entry = DirectoryEntry(source_hash, display_name, trust_level, preferred_delivery=delivery, sort_rank=sort_rank, notes=notes_value) self.app.directory.remember(entry) self._refresh_open_conversation_widget(source_hash_text) self.dialog_open = False self.update_conversation_list() self.app.ui.main_display.sub_displays.network_display.directory_change_callback() except Exception as e: RNS.log("Could not save directory entry. The contained exception was: "+str(e), RNS.LOG_VERBOSE) if not dialog_pile.error_display: dialog_pile.error_display = True options = dialog_pile.options(height_type=urwid.PACK) dialog_pile.contents.append((urwid.Text(""), options)) dialog_pile.contents.append(( urwid.Text(("error_text", "Could not save entry. Check your input."), align=urwid.CENTER), options,) ) source_is_known = self.app.directory.is_known(bytes.fromhex(source_hash_text)) if source_is_known: known_section = urwid.Divider(g["divider1"]) else: def query_action(sender, user_data): self.close_conversation_by_hash(user_data) nomadnet.Conversation.query_for_peer(user_data) options = dialog_pile.options(height_type=urwid.PACK) dialog_pile.contents = [ (urwid.Text("Query sent"), options), (urwid.Button("OK", on_press=dismiss_dialog), options) ] query_button = urwid.Button("Query network for keys", on_press=query_action, user_data=source_hash_text) known_section = urwid.Pile([ urwid.Divider(g["divider1"]), urwid.Text(g["info"]+"\n", align=urwid.CENTER), urwid.Text( "The identity of this peer is not known, and you cannot currently send messages to it. " "You can query the network to obtain the identity.\n", align=urwid.CENTER, ), query_button, urwid.Divider(g["divider1"]), ]) action_status = urwid.Text("", align=urwid.CENTER) ping_button = urwid.Button("Ping") block_button = urwid.Button("Block") qr_button = urwid.Button("LXMF") urwid.connect_signal(ping_button, "click", lambda _b: self._ping_peer_from_dialog(source_hash_text, action_status, ping_button)) urwid.connect_signal(block_button, "click", lambda _b: self._block_peer_from_dialog(source_hash_text)) urwid.connect_signal(qr_button, "click", lambda _b: self.show_qr_dialog(source_hash_text, title=display_name or source_hash_text)) actions_row = urwid.Columns([ (urwid.WEIGHT, 0.32, ping_button), (urwid.WEIGHT, 0.02, urwid.Text("")), (urwid.WEIGHT, 0.32, block_button), (urwid.WEIGHT, 0.02, urwid.Text("")), (urwid.WEIGHT, 0.32, qr_button), ]) dialog_pile = urwid.Pile([ selected_id_widget, e_name, e_copy, urwid.Divider(g["divider1"]), r_untrusted, r_unknown, r_trusted, urwid.Divider(g["divider1"]), r_direct, r_propagated, urwid.Divider(g["divider1"]), cb_pin, e_notes, known_section, actions_row, action_status, urwid.Divider(g["divider1"]), urwid.Columns([ (urwid.WEIGHT, 0.45, urwid.Button("Save", on_press=confirmed)), (urwid.WEIGHT, 0.1, urwid.Text("")), (urwid.WEIGHT, 0.45, urwid.Button("Back", on_press=dismiss_dialog)), ]) ]) dialog_pile.error_display = False dialog = DialogLineBox(dialog_pile, title="Peer Info") dialog.delegate = self bottom = self.listbox overlay = urwid.Overlay( dialog, bottom, align=urwid.CENTER, width=urwid.RELATIVE_100, valign=urwid.MIDDLE, height=urwid.PACK, left=2, right=2, ) # options = self.columns_widget.options(urwid.WEIGHT, ConversationsDisplay.list_width) options = self.columns_widget.options(urwid.GIVEN, ConversationsDisplay.given_list_width) self.columns_widget.contents[0] = (overlay, options) def new_conversation(self): self.dialog_open = True source_hash = "" display_name = "" e_id = urwid.Edit(caption="Addr : ",edit_text=source_hash) e_name = urwid.Edit(caption="Name : ",edit_text=display_name) trust_button_group = [] r_untrusted = urwid.RadioButton(trust_button_group, "Untrusted") r_unknown = urwid.RadioButton(trust_button_group, "Unknown", state=True) r_trusted = urwid.RadioButton(trust_button_group, "Trusted") def dismiss_dialog(sender): self.update_conversation_list() self.dialog_open = False def confirmed(sender): try: existing_conversations = nomadnet.Conversation.conversation_list(self.app) display_name = e_name.get_edit_text() source_hash_text = e_id.get_edit_text().strip() source_hash = bytes.fromhex(source_hash_text) trust_level = DirectoryEntry.UNTRUSTED if r_unknown.state == True: trust_level = DirectoryEntry.UNKNOWN elif r_trusted.state == True: trust_level = DirectoryEntry.TRUSTED if not source_hash in [c[0] for c in existing_conversations]: entry = DirectoryEntry(source_hash, display_name, trust_level) self.app.directory.remember(entry) new_conversation = nomadnet.Conversation(source_hash_text, nomadnet.NomadNetworkApp.get_shared_instance(), initiator=True) self.update_conversation_list() if trust_level != DirectoryEntry.TRUSTED: if self.list_filter != ConversationsDisplay.LIST_FILTER_UNTRUSTED: self._set_filter(ConversationsDisplay.LIST_FILTER_UNTRUSTED) self.display_conversation(source_hash_text) self.dialog_open = False except Exception as e: RNS.log("Could not start conversation. The contained exception was: "+str(e), RNS.LOG_VERBOSE) if not dialog_pile.error_display: dialog_pile.error_display = True options = dialog_pile.options(height_type=urwid.PACK) dialog_pile.contents.append((urwid.Text(""), options)) dialog_pile.contents.append(( urwid.Text( ("error_text", "Could not start conversation. Check your input."), align=urwid.CENTER, ), options, )) dialog_pile = urwid.Pile([ e_id, e_name, urwid.Text(""), r_untrusted, r_unknown, r_trusted, urwid.Text(""), urwid.Columns([ (urwid.WEIGHT, 0.45, urwid.Button("Create", on_press=confirmed)), (urwid.WEIGHT, 0.1, urwid.Text("")), (urwid.WEIGHT, 0.45, urwid.Button("Back", on_press=dismiss_dialog)), ]) ]) dialog_pile.error_display = False dialog = DialogLineBox(dialog_pile, title="New Conversation") dialog.delegate = self bottom = self.listbox overlay = urwid.Overlay( dialog, bottom, align=urwid.CENTER, width=urwid.RELATIVE_100, valign=urwid.MIDDLE, height=urwid.PACK, left=2, right=2, ) # options = self.columns_widget.options(urwid.WEIGHT, ConversationsDisplay.list_width) options = self.columns_widget.options(urwid.GIVEN, ConversationsDisplay.given_list_width) self.columns_widget.contents[0] = (overlay, options) def ingest_lxm_uri(self): self.dialog_open = True lxm_uri = "" e_uri = urwid.Edit(caption="URI : ",edit_text=lxm_uri) def dismiss_dialog(sender): self.update_conversation_list() self.dialog_open = False def confirmed(sender): try: local_delivery_signal = "local_delivery_occurred" duplicate_signal = "duplicate_lxm" lxm_uri = e_uri.get_edit_text().strip() ingest_result = self.app.message_router.ingest_lxm_uri( lxm_uri, signal_local_delivery=local_delivery_signal, signal_duplicate=duplicate_signal ) if ingest_result == False: raise ValueError("The URI contained no decodable messages") elif ingest_result == local_delivery_signal: rdialog_pile = urwid.Pile([ urwid.Text("Message was decoded, decrypted successfully, and added to your conversation list."), urwid.Text(""), urwid.Columns([ (urwid.WEIGHT, 0.6, urwid.Text("")), (urwid.WEIGHT, 0.4, urwid.Button("OK", on_press=dismiss_dialog)), ]) ]) rdialog_pile.error_display = False rdialog = DialogLineBox(rdialog_pile, title="Ingest message URI") rdialog.delegate = self bottom = self.listbox roverlay = urwid.Overlay( rdialog, bottom, align=urwid.CENTER, width=urwid.RELATIVE_100, valign=urwid.MIDDLE, height=urwid.PACK, left=2, right=2, ) options = self.columns_widget.options(urwid.GIVEN, ConversationsDisplay.given_list_width) self.columns_widget.contents[0] = (roverlay, options) elif ingest_result == duplicate_signal: rdialog_pile = urwid.Pile([ urwid.Text("The decoded message has already been processed by the LXMF Router, and will not be ingested again."), urwid.Text(""), urwid.Columns([ (urwid.WEIGHT, 0.6, urwid.Text("")), (urwid.WEIGHT, 0.4, urwid.Button("OK", on_press=dismiss_dialog)), ]) ]) rdialog_pile.error_display = False rdialog = DialogLineBox(rdialog_pile, title="Ingest message URI") rdialog.delegate = self bottom = self.listbox roverlay = urwid.Overlay( rdialog, bottom, align=urwid.CENTER, width=urwid.RELATIVE_100, valign=urwid.MIDDLE, height=urwid.PACK, left=2, right=2, ) options = self.columns_widget.options(urwid.GIVEN, ConversationsDisplay.given_list_width) self.columns_widget.contents[0] = (roverlay, options) else: if self.app.enable_node: propagation_text = "The decoded message was not addressed to this LXMF address, but has been added to the propagation node queues, and will be distributed on the propagation network." else: propagation_text = "The decoded message was not addressed to this LXMF address, and has been discarded." rdialog_pile = urwid.Pile([ urwid.Text(propagation_text), urwid.Text(""), urwid.Columns([ (urwid.WEIGHT, 0.6, urwid.Text("")), (urwid.WEIGHT, 0.4, urwid.Button("OK", on_press=dismiss_dialog)), ]) ]) rdialog_pile.error_display = False rdialog = DialogLineBox(rdialog_pile, title="Ingest message URI") rdialog.delegate = self bottom = self.listbox roverlay = urwid.Overlay( rdialog, bottom, align=urwid.CENTER, width=urwid.RELATIVE_100, valign=urwid.MIDDLE, height=urwid.PACK, left=2, right=2, ) options = self.columns_widget.options(urwid.GIVEN, ConversationsDisplay.given_list_width) self.columns_widget.contents[0] = (roverlay, options) except Exception as e: RNS.log("Could not ingest LXM URI. The contained exception was: "+str(e), RNS.LOG_VERBOSE) if not dialog_pile.error_display: dialog_pile.error_display = True options = dialog_pile.options(height_type=urwid.PACK) dialog_pile.contents.append((urwid.Text(""), options)) dialog_pile.contents.append((urwid.Text(("error_text", "Could ingest LXM from URI data. Check your input."), align=urwid.CENTER), options)) dialog_pile = urwid.Pile([ e_uri, urwid.Text(""), urwid.Columns([ (urwid.WEIGHT, 0.45, urwid.Button("Ingest", on_press=confirmed)), (urwid.WEIGHT, 0.1, urwid.Text("")), (urwid.WEIGHT, 0.45, urwid.Button("Back", on_press=dismiss_dialog)), ]) ]) dialog_pile.error_display = False dialog = DialogLineBox(dialog_pile, title="Ingest message URI") dialog.delegate = self bottom = self.listbox overlay = urwid.Overlay( dialog, bottom, align=urwid.CENTER, width=urwid.RELATIVE_100, valign=urwid.MIDDLE, height=urwid.PACK, left=2, right=2, ) options = self.columns_widget.options(urwid.GIVEN, ConversationsDisplay.given_list_width) self.columns_widget.contents[0] = (overlay, options) def delete_conversation(self, source_hash): if source_hash in ConversationsDisplay.cached_conversation_widgets: conversation = ConversationsDisplay.cached_conversation_widgets[source_hash] self.close_conversation(conversation) def toggle_fullscreen(self): if ConversationsDisplay.given_list_width != 0: self.saved_list_width = ConversationsDisplay.given_list_width ConversationsDisplay.given_list_width = 0 else: ConversationsDisplay.given_list_width = self.saved_list_width self.update_conversation_list() def _decode_pn_app_data(self, app_data): try: if not pn_announce_data_is_valid(app_data): return None data = msgpack.unpackb(app_data) enabled = bool(data[2]) name = None try: meta = data[6] if isinstance(meta, dict) and PN_META_NAME in meta: raw = meta[PN_META_NAME] if isinstance(raw, (bytes, bytearray)): name = raw.decode("utf-8", errors="replace") except Exception: name = None return {"enabled": enabled, "name": name} except Exception: return None def _pn_dropdown_label(self, pn_hash, meta): name = (meta or {}).get("name") if name: label = strip_modifiers(name) or "" label = " ".join(label.split()) else: label = "" if not label: label = RNS.prettyhexrep(pn_hash) max_len = 40 if len(label) > max_len: label = label[:max_len-1]+"…" tail = "<"+RNS.hexrep(pn_hash, delimit=False)+">" if tail not in label: label = label+" "+tail if meta is None: status = "[?]" elif meta.get("enabled"): status = "[E]" else: status = "[D]" return status+" "+label def _build_pn_options(self): options = [] seen = set() try: pn_announces = list(self.app.directory._pn_announces) except Exception: pn_announces = [] for tup in pn_announces: if len(tup) < 3: continue pn_hash = tup[1] app_data = tup[2] if pn_hash in seen: continue seen.add(pn_hash) meta = self._decode_pn_app_data(app_data) options.append((pn_hash, self._pn_dropdown_label(pn_hash, meta))) for extra in (self.app.get_user_selected_propagation_node(), self.app.get_default_propagation_node()): if extra is None or extra in seen: continue seen.add(extra) meta = None try: cached = RNS.Identity.recall_app_data(extra) if cached is not None: meta = self._decode_pn_app_data(cached) except Exception: meta = None options.append((extra, self._pn_dropdown_label(extra, meta))) return options def sync_conversations(self): g = self.app.ui.glyphs self.dialog_open = True def dismiss_dialog(sender): self.dialog_open = False self.sync_dialog = None self.update_conversation_list() if self.app.message_router.propagation_transfer_state >= LXMF.LXMRouter.PR_COMPLETE: self.app.cancel_lxmf_sync() max_messages_group = [] r_mall = urwid.RadioButton(max_messages_group, "Download all", state=True) r_mlim = urwid.RadioButton(max_messages_group, "Limit to", state=False) ie_lim = urwid.IntEdit("", 5) rbs = urwid.GridFlow([r_mlim, ie_lim], 12, 1, 0, align=urwid.LEFT) def sync_now(sender): limit = None if r_mlim.get_state(): limit = ie_lim.value() self.app.request_lxmf_sync(limit) self.update_sync_dialog() def cancel_sync(sender): self.app.cancel_lxmf_sync() self.update_sync_dialog() cancel_button = urwid.Button("Close", on_press=dismiss_dialog) sync_progress = SyncProgressBar("progress_empty" , "progress_full", current=self.app.get_sync_progress(), done=1.0, satt=None) real_sync_button = urwid.Button("Sync Now", on_press=sync_now) hidden_sync_button = urwid.Button("Cancel Sync", on_press=cancel_sync) if self.app.get_sync_status() == "Idle" or self.app.message_router.propagation_transfer_state >= LXMF.LXMRouter.PR_COMPLETE: sync_button = real_sync_button else: sync_button = hidden_sync_button button_columns = urwid.Columns([ (urwid.WEIGHT, 0.45, sync_button), (urwid.WEIGHT, 0.1, urwid.Text("")), (urwid.WEIGHT, 0.45, cancel_button), ]) real_sync_button.bc = button_columns current_default = self.app.get_default_propagation_node() user_selected = self.app.get_user_selected_propagation_node() pn_options = self._build_pn_options() selected_target = user_selected if user_selected is not None else current_default def on_pn_picked(picked_hash): try: self.app.set_user_selected_propagation_node(picked_hash) except Exception as e: RNS.log("Could not update propagation node: "+str(e), RNS.LOG_ERROR) def show_set_pn_dialog(_sender): current_pn = self.app.get_user_selected_propagation_node() current_str = RNS.hexrep(current_pn, delimit=False) if current_pn is not None else "" pn_edit = urwid.Edit(caption="Hash : ", edit_text=current_str) status_text = urwid.Text("", align=urwid.CENTER) def reopen_sync(_b=None): self.sync_conversations() def save_pn(_b): text = pn_edit.get_edit_text().strip().replace(":", "").replace(" ", "") expected_len = RNS.Reticulum.TRUNCATED_HASHLENGTH // 8 if text == "": self.app.set_user_selected_propagation_node(None) else: try: node_hash = bytes.fromhex(text) except ValueError: status_text.set_text("Invalid hex") return if len(node_hash) != expected_len: status_text.set_text("Must be "+str(expected_len)+" bytes ("+str(expected_len*2)+" hex chars)") return self.app.set_user_selected_propagation_node(node_hash) reopen_sync() def clear_pn(_b): pn_edit.set_edit_text("") self.app.set_user_selected_propagation_node(None) reopen_sync() inner = DialogLineBox( urwid.Pile([ urwid.Text("Enter an LXMF propagation\ndestination hash as hex.", align=urwid.CENTER), urwid.Divider(), pn_edit, urwid.Divider(), status_text, urwid.Columns([ (urwid.WEIGHT, 0.3, urwid.Button("Save", on_press=save_pn)), (urwid.WEIGHT, 0.05, urwid.Text("")), (urwid.WEIGHT, 0.3, urwid.Button("Clear", on_press=clear_pn)), (urwid.WEIGHT, 0.05, urwid.Text("")), (urwid.WEIGHT, 0.3, urwid.Button("Close", on_press=reopen_sync)), ]) ]), title="Set Propagation Node", ) inner.delegate = self self.sync_dialog = None overlay = urwid.Overlay( inner, self.listbox, align=urwid.CENTER, width=urwid.RELATIVE_100, valign=urwid.MIDDLE, height=urwid.PACK, left=2, right=2, ) options = self.columns_widget.options(urwid.GIVEN, ConversationsDisplay.given_list_width) self.columns_widget.contents[0] = (overlay, options) set_pn_button = urwid.Button("Custom Node...", on_press=show_set_pn_dialog) if pn_options: node_picker = PropNodePicker(pn_options, selected_target, on_pn_picked) node_selector = urwid.Pile([ urwid.Text("Propagation node:"), node_picker, set_pn_button, ]) else: node_selector = None pn_ident = None if current_default is not None: pn_ident = RNS.Identity.recall(current_default) if pn_ident is None: RNS.log("Propagation node identity is unknown, requesting from network...", RNS.LOG_DEBUG) RNS.Transport.request_path(current_default) if pn_ident is not None or node_selector is not None: header_str = "" if pn_ident is not None: node_hash = RNS.Destination.hash_from_name_and_identity("nomadnetwork.node", pn_ident) pn_entry = self.app.directory.find(node_hash) if pn_entry is not None and getattr(pn_entry, "display_name", None): header_str = " "+strip_modifiers(str(pn_entry.display_name)) else: header_str = " "+RNS.prettyhexrep(current_default) else: header_str = " (no default)" pile_items = [ urwid.Text(""+g["node"]+header_str, align=urwid.CENTER), urwid.Divider(g["divider1"]), sync_progress, urwid.Divider(g["divider1"]), ] if node_selector is not None: pile_items += [node_selector, urwid.Divider(g["divider1"])] pile_items += [r_mall, rbs, urwid.Text(""), button_columns] dialog = DialogLineBox(urwid.Pile(pile_items), title="Message Sync") else: button_columns = urwid.Columns([ (urwid.WEIGHT, 0.45, urwid.Text("" )), (urwid.WEIGHT, 0.1, urwid.Text("")), (urwid.WEIGHT, 0.45, cancel_button), ]) dialog = DialogLineBox( urwid.Pile([ urwid.Text(""), urwid.Text("No trusted nodes found, cannot sync!\n", align=urwid.CENTER), urwid.Text( "To synchronise messages from the network, " "one or more nodes must be marked as trusted in the Known Nodes list, " "or a node must manually be selected as the default propagation node. " "Nomad Network will then automatically sync from the nearest trusted node, " "or the manually selected one.", align=urwid.LEFT, ), urwid.Text(""), button_columns ]), title="Message Sync" ) dialog.delegate = self dialog.sync_progress = sync_progress dialog.cancel_button = cancel_button dialog.real_sync_button = real_sync_button dialog.hidden_sync_button = hidden_sync_button dialog.bc = button_columns self.sync_dialog = dialog bottom = self.listbox overlay = urwid.Overlay( dialog, bottom, align=urwid.CENTER, width=urwid.RELATIVE_100, valign=urwid.MIDDLE, height=urwid.PACK, left=2, right=2, ) # options = self.columns_widget.options(urwid.WEIGHT, ConversationsDisplay.list_width) options = self.columns_widget.options(urwid.GIVEN, ConversationsDisplay.given_list_width) self.columns_widget.contents[0] = (overlay, options) def update_sync_dialog(self, loop = None, sender = None): if self.dialog_open and self.sync_dialog != None: self.sync_dialog.sync_progress.set_completion(self.app.get_sync_progress()) if self.app.get_sync_status() == "Idle" or self.app.message_router.propagation_transfer_state >= LXMF.LXMRouter.PR_COMPLETE: self.sync_dialog.bc.contents[0] = (self.sync_dialog.real_sync_button, self.sync_dialog.bc.options(urwid.WEIGHT, 0.45)) else: self.sync_dialog.bc.contents[0] = (self.sync_dialog.hidden_sync_button, self.sync_dialog.bc.options(urwid.WEIGHT, 0.45)) self.app.ui.loop.set_alarm_in(0.2, self.update_sync_dialog) def conversation_list_selection(self, arg1, arg2): pass def update_conversation_list(self): selected_hash = None selected_item = self.ilb.get_selected_item() if selected_item is not None: if hasattr(selected_item, "source_hash"): selected_hash = selected_item.source_hash self.update_listbox() options = self.columns_widget.options(urwid.GIVEN, ConversationsDisplay.given_list_width) if not self.dialog_open: self.columns_widget.contents[0] = (self.listbox, options) elif self.sync_dialog is not None: bottom = self.listbox overlay = urwid.Overlay( self.sync_dialog, bottom, align=urwid.CENTER, width=urwid.RELATIVE_100, valign=urwid.MIDDLE, height=urwid.PACK, left=2, right=2, ) self.columns_widget.contents[0] = (overlay, options) # else: another dialog (peer info, new conversation, block confirm, etc.) is # open as an overlay in contents[0]; leave it alone so an incoming message # doesn't dismiss it. The underlying listbox is a persistent widget and was # already refreshed by update_listbox() above. if selected_hash is not None: for idx, widget in enumerate(self.list_widgets): if widget.source_hash == selected_hash: self.ilb.select_item(idx) break nomadnet.NomadNetworkApp.get_shared_instance().ui.loop.draw_screen() if self.app.ui.main_display.sub_displays.active_display == self.app.ui.main_display.sub_displays.conversations_display: if self.currently_displayed_conversation != None: if self.app.conversation_is_unread(self.currently_displayed_conversation): self.app.mark_conversation_read(self.currently_displayed_conversation) try: if os.path.isfile(self.app.conversationpath + "/" + self.currently_displayed_conversation + "/unread"): os.unlink(self.app.conversationpath + "/" + self.currently_displayed_conversation + "/unread") except Exception as e: raise e def display_conversation(self, sender=None, source_hash=None): if self.currently_displayed_conversation != None: if self.app.conversation_is_unread(self.currently_displayed_conversation): self.app.mark_conversation_read(self.currently_displayed_conversation) self.currently_displayed_conversation = source_hash options = self.columns_widget.options(urwid.WEIGHT, 1) self.columns_widget.contents[1] = (self.make_conversation_widget(source_hash), options) if source_hash == None: self.columns_widget.focus_position = 0 else: if self.app.conversation_is_unread(source_hash): self.app.mark_conversation_read(source_hash) self.update_conversation_list() self.columns_widget.focus_position = 1 conversation_position = None index = 0 for widget in self.list_widgets: if widget.source_hash == source_hash: conversation_position = index index += 1 if conversation_position != None: self.ilb.select_item(conversation_position) def make_conversation_widget(self, source_hash): if source_hash in ConversationsDisplay.cached_conversation_widgets: conversation_widget = ConversationsDisplay.cached_conversation_widgets[source_hash] if source_hash != None: conversation_widget.update_message_widgets(replace=True) conversation_widget.check_editor_allowed() return conversation_widget else: widget = ConversationWidget(source_hash, delegate=self) ConversationsDisplay.cached_conversation_widgets[source_hash] = widget widget.check_editor_allowed() return widget def close_conversation_by_hash(self, conversation_hash): if conversation_hash in ConversationsDisplay.cached_conversation_widgets: ConversationsDisplay.cached_conversation_widgets.pop(conversation_hash) if self.currently_displayed_conversation == conversation_hash: self.display_conversation(sender=None, source_hash=None) def close_conversation(self, conversation): if conversation.source_hash in ConversationsDisplay.cached_conversation_widgets: ConversationsDisplay.cached_conversation_widgets.pop(conversation.source_hash) if self.currently_displayed_conversation == conversation.source_hash: self.display_conversation(sender=None, source_hash=None) def conversation_list_widget(self, conversation): trust_level = conversation[2] display_name = conversation[1] source_hash = conversation[0] unread = conversation[4] last_activity = conversation[5] failed = conversation[6] if len(conversation) > 6 else 0 g = self.app.ui.glyphs if trust_level == DirectoryEntry.UNTRUSTED: symbol = g["cross"] style = "list_untrusted" focus_style = "list_focus_untrusted" elif trust_level == DirectoryEntry.UNKNOWN: symbol = "?" style = "list_unknown" focus_style = "list_focus" elif trust_level == DirectoryEntry.TRUSTED: symbol = g["check"] style = "list_normal" focus_style = "list_focus" elif trust_level == DirectoryEntry.WARNING: symbol = g["warning"] style = "list_warning" focus_style = "list_focus" else: symbol = g["warning"] style = "list_untrusted" focus_style = "list_focus_untrusted" is_pinned = False try: entry = self.app.directory.find(bytes.fromhex(source_hash)) is_pinned = entry is not None and entry.sort_rank is not None except Exception: is_pinned = False head = symbol if is_pinned: head = g.get("pin", "*") + " " + head if display_name != None and display_name != "": head += " "+display_name if trust_level != DirectoryEntry.TRUSTED: head += " <"+source_hash+">" markup = [head] if failed and source_hash != self.currently_displayed_conversation: badge_text = " "+g["warning"]+" ("+str(failed)+")" # markup.append(("msg_notice_caution", badge_text)) markup.append(badge_text) elif unread and source_hash != self.currently_displayed_conversation: badge_text = " "+g["unread"]+" ("+str(unread)+")" # Good idea with having the badges here colored, but # using the bg color for it is a bit much, I think. # I set fg color attrmap styles, but that messes up # the bg on list focus. If there's a way to handle # that, we can re-enable this. # markup.append(("msg_notice_unread", badge_text)) markup.append(badge_text) if trust_level == DirectoryEntry.TRUSTED and unread and source_hash != self.currently_displayed_conversation: style = "msg_notice_unread" if last_activity > 0: markup.append("\n "+relative_time(last_activity)) widget = ListEntry(markup) urwid.connect_signal(widget, "click", self.display_conversation, conversation[0]) display_widget = urwid.AttrMap(widget, style, focus_style) display_widget.source_hash = source_hash display_widget.display_name = display_name return display_widget def shortcuts(self): try: focus_path = self.columns_widget.get_focus_path() except Exception: return self.list_shortcuts if not focus_path: return self.list_shortcuts if focus_path[0] == 0: return self.list_shortcuts elif focus_path[0] == 1: return self.editor_shortcuts else: return self.list_shortcuts class ListEntry(urwid.Text): _selectable = True signals = ["click"] def keypress(self, size, key): """ Send 'click' signal on 'activate' command. """ if self._command_map[key] != urwid.ACTIVATE: return key self._emit('click') def mouse_event(self, size, event, button, x, y, focus): """ Send 'click' signal on button 1 press. """ if button != 1 or not urwid.util.is_mouse_press(event): return False self._emit('click') return True class MessageEdit(urwid.Edit): def keypress(self, size, key): if key == "ctrl d": self.delegate.send_message() elif key == "ctrl p": self.delegate.paper_message() elif key == "ctrl a": self.delegate.attach_file() elif key == "ctrl s": self.delegate.save_focused_attachments() elif key == "ctrl k": self.delegate.clear_editor() elif key == "up": y = self.get_cursor_coords(size)[1] if y == 0: if self.delegate.full_editor_active and self.name == "title_editor": self.delegate.frame.focus_position = "body" elif not self.delegate.full_editor_active and self.name == "content_editor": self.delegate.frame.focus_position = "body" else: return super(MessageEdit, self).keypress(size, key) else: return super(MessageEdit, self).keypress(size, key) else: return super(MessageEdit, self).keypress(size, key) class ConversationFrame(urwid.Frame): def keypress(self, size, key): if self.focus_position == "header": result = super(ConversationFrame, self).keypress(size, key) if result == "up": nomadnet.NomadNetworkApp.get_shared_instance().ui.main_display.frame.focus_position = "header" return None if result == "down": self.focus_position = "body" return None return result if self.focus_position == "body": if getattr(self.delegate, "dialog_active", False) or getattr(self.delegate, "dialog_open", False): return super(ConversationFrame, self).keypress(size, key) elif key == "up" and self.delegate.messagelist.top_is_visible: if getattr(self.delegate, "has_visible_trust_banner", lambda: False)(): try: self.delegate._header_pile.focus_position = 1 self.focus_position = "header" return None except Exception: pass nomadnet.NomadNetworkApp.get_shared_instance().ui.main_display.frame.focus_position = "header" elif key == "down" and self.delegate.messagelist.bottom_is_visible: self.focus_position = "footer" else: return super(ConversationFrame, self).keypress(size, key) elif key == "ctrl k": self.delegate.clear_editor() else: return super(ConversationFrame, self).keypress(size, key) class ConversationWidget(urwid.WidgetWrap): def __init__(self, source_hash, delegate): self.app = nomadnet.NomadNetworkApp.get_shared_instance() g = self.app.ui.glyphs self.delegate = delegate if source_hash == None: self.frame = None display_widget = urwid.LineBox(urwid.Filler(urwid.Text("\n No conversation selected"), "top")) super().__init__(display_widget) else: if source_hash in ConversationsDisplay.cached_conversation_widgets: return ConversationsDisplay.cached_conversation_widgets[source_hash] else: self.source_hash = source_hash self.conversation = nomadnet.Conversation(source_hash, nomadnet.NomadNetworkApp.get_shared_instance()) self.message_widgets = [] self.sort_by_timestamp = False self.pending_attachments = [] self.dialog_active = False self.update_message_widgets() self.conversation.register_changed_callback(self._on_conversation_changed_from_callback) #title_editor = MessageEdit(caption="\u270E", edit_text="", multiline=False) title_editor = MessageEdit(caption="", edit_text="", multiline=False) title_editor.delegate = self title_editor.name = "title_editor" #msg_editor = MessageEdit(caption="\u270E", edit_text="", multiline=True) msg_editor = MessageEdit(caption="", edit_text="", multiline=True) msg_editor.delegate = self msg_editor.name = "content_editor" self.peer_info_widget = urwid.AttrMap(urwid.Text(""), "msg_header_sent") self._update_peer_info() self._trust_banner_dismissed = False self._header_pile = urwid.Pile([self.peer_info_widget]) self._refresh_trust_banner() header = self._header_pile self.minimal_editor = urwid.AttrMap(msg_editor, "msg_editor") self.minimal_editor.name = "minimal_editor" title_columns = urwid.Columns([ (8, urwid.Text("Title")), urwid.AttrMap(title_editor, "msg_editor"), ]) content_columns = urwid.Columns([ (8, urwid.Text("Content")), urwid.AttrMap(msg_editor, "msg_editor") ]) self.full_editor = urwid.Pile([ title_columns, content_columns ]) self.full_editor.name = "full_editor" self.content_editor = msg_editor self.title_editor = title_editor self.full_editor_active = False self.frame = ConversationFrame( self.messagelist, header=header, footer=self.minimal_editor, focus_part="footer" ) self.frame.delegate = self self.display_widget = urwid.LineBox( self.frame ) super().__init__(self.display_widget) def has_visible_trust_banner(self): if self._trust_banner_dismissed: return False try: tl = self.app.directory.trust_level(bytes.fromhex(self.source_hash)) except Exception: tl = DirectoryEntry.UNKNOWN return tl != DirectoryEntry.TRUSTED def _refresh_trust_banner(self): contents = [(self.peer_info_widget, self._header_pile.options())] if self.has_visible_trust_banner(): banner = self._build_trust_banner() contents.append((banner, self._header_pile.options())) self._header_pile.contents = contents if len(contents) > 1: try: self._header_pile.focus_position = 1 except Exception: pass def _build_trust_banner(self): g = self.app.ui.glyphs msg = urwid.Text(" "+g["warning"]+" This peer isn't trusted yet.") btn_trust = urwid.Button("Trust", on_press=self._on_trust_click) btn_block = urwid.Button("Block", on_press=self._on_block_click) btn_nothing = urwid.Button("Do nothing", on_press=self._on_ignore_click) row = urwid.Columns([ ('weight', 1, msg), ('pack', btn_trust), (1, urwid.Text(" ")), ('pack', btn_block), (1, urwid.Text(" ")), ('pack', btn_nothing), (1, urwid.Text(" ")), ], dividechars=0) return urwid.AttrMap(row, "msg_warning_untrusted") def _on_trust_click(self, _btn): try: src = bytes.fromhex(self.source_hash) existing = self.app.directory.find(src) display_name = getattr(existing, "display_name", None) if existing is not None else None preferred = getattr(existing, "preferred_delivery", None) if existing is not None else None entry = DirectoryEntry(src, display_name, DirectoryEntry.TRUSTED, preferred_delivery=preferred) self.app.directory.remember(entry) except Exception as e: RNS.log("Could not mark peer as trusted: "+str(e), RNS.LOG_ERROR) self._refresh_trust_banner() try: self.frame.focus_position = "footer" except Exception: pass try: if self.delegate.list_filter != ConversationsDisplay.LIST_FILTER_TRUSTED: self.delegate._set_filter(ConversationsDisplay.LIST_FILTER_TRUSTED) else: self.delegate.update_conversation_list() except Exception as e: RNS.log("Trust UI refresh failed: "+str(e), RNS.LOG_ERROR) def _on_ignore_click(self, _btn): self._trust_banner_dismissed = True self._refresh_trust_banner() def _on_block_click(self, _btn): def dismiss(_b): self.dialog_active = False try: self.delegate.dialog_open = False except Exception: pass try: self.delegate.update_conversation_list() except Exception: pass def confirmed(_b): self.dialog_active = False try: self.delegate.dialog_open = False except Exception: pass try: self._block_peer() except Exception as e: RNS.log("Block failed: "+str(e), RNS.LOG_ERROR) try: self.delegate.update_conversation_list() except Exception: pass try: who = self.app.directory.simplest_display_str(bytes.fromhex(self.source_hash)) except Exception: who = self.source_hash dialog = DialogLineBox( urwid.Pile([ urwid.Text(""), urwid.Text("Block "+str(who)+"?\n\nThis will blackhole the peer's identity in Reticulum,\nadd them to your ignored list, and delete this conversation.\n", align=urwid.CENTER), urwid.Columns([ (urwid.WEIGHT, 0.45, urwid.Button("Yes, block", on_press=confirmed)), (urwid.WEIGHT, 0.10, urwid.Text("")), (urwid.WEIGHT, 0.45, urwid.Button("Cancel", on_press=dismiss)), ]), ]), title="Confirm block" ) dialog.delegate = self.delegate bottom = self.delegate.listbox overlay = urwid.Overlay(dialog, bottom, align=urwid.CENTER, width=urwid.RELATIVE_100, valign=urwid.MIDDLE, height=urwid.PACK, left=2, right=2) try: self.delegate.columns_widget.contents[0] = ( overlay, self.delegate.columns_widget.options(urwid.GIVEN, ConversationsDisplay.given_list_width), ) self.delegate.columns_widget.focus_position = 0 self.dialog_active = True try: self.delegate.dialog_open = True except Exception: pass except Exception: pass def _block_peer(self): try: src = bytes.fromhex(self.source_hash) except Exception: return try: self.app.block_destination(src, reason="user-blocked from nomadnet conversation") except Exception as e: RNS.log("Block failed: "+str(e), RNS.LOG_ERROR) try: self.delegate.delete_conversation(self.source_hash) nomadnet.Conversation.delete_conversation(self.source_hash, self.app) except Exception as e: RNS.log("Could not delete blocked conversation: "+str(e), RNS.LOG_ERROR) def _update_peer_info(self): def san(name): if self.app.config["textui"]["sanitize_names"]: return sanitize_name(name) else: return strip_modifiers(name) g = self.app.ui.glyphs source_hash_bytes = bytes.fromhex(self.source_hash) display_name = self.app.directory.display_name(source_hash_bytes) app_data = None if display_name is None or self.app.message_router.get_outbound_stamp_cost(source_hash_bytes) is None: app_data = RNS.Identity.recall_app_data(source_hash_bytes) if display_name is None: if app_data: display_name = san(LXMF.display_name_from_app_data(app_data)) if display_name is None: display_name = RNS.prettyhexrep(source_hash_bytes) stamp_cost = self.app.message_router.get_outbound_stamp_cost(source_hash_bytes) if stamp_cost is None and app_data: stamp_cost = LXMF.stamp_cost_from_app_data(app_data) hops = RNS.Transport.hops_to(source_hash_bytes) if hops >= RNS.Transport.PATHFINDER_M: hops_str = "unknown" else: hops_str = str(hops)+" hop" + ("s" if hops != 1 else "") right_parts = [] if stamp_cost is not None: right_parts.append("Stamp: "+str(stamp_cost)) right_parts.append(g["speed"]+hops_str) left = " "+display_name right = " ".join(right_parts)+" " self.peer_info_widget.original_widget.set_text(left+" | "+right) def clear_history_dialog(self): def dismiss_dialog(sender): self.dialog_open = False self.conversation_changed(None) def confirmed(sender): self.dialog_open = False self.conversation.clear_history() self.conversation_changed(None) dialog = DialogLineBox( urwid.Pile([ urwid.Text("Clear conversation history\n", align=urwid.CENTER), urwid.Columns([ (urwid.WEIGHT, 0.45, urwid.Button("Yes", on_press=confirmed)), (urwid.WEIGHT, 0.1, urwid.Text("")), (urwid.WEIGHT, 0.45, urwid.Button("No", on_press=dismiss_dialog)), ]) ]), title="?" ) dialog.delegate = self bottom = self.messagelist overlay = urwid.Overlay( dialog, bottom, align=urwid.CENTER, width=34, valign=urwid.MIDDLE, height=urwid.PACK, left=2, right=2, ) self.frame.contents["body"] = (overlay, self.frame.options()) self.frame.focus_position = "body" def _build_footer(self): g = self.app.ui.glyphs if self.full_editor_active: editor = self.full_editor else: editor = self.minimal_editor if self.pending_attachments: attachment_texts = [] for path in self.pending_attachments: attachment_texts.append(os.path.basename(path)) indicator = urwid.AttrMap( urwid.Text(g["file"]+" "+str(len(self.pending_attachments))+" file(s): "+", ".join(attachment_texts)), "msg_header_sent", ) return urwid.Pile([indicator, editor]) else: return editor def toggle_editor(self): if self.full_editor_active: self.full_editor_active = False else: self.full_editor_active = True self.frame.contents["footer"] = (self._build_footer(), None) def check_editor_allowed(self): g = self.app.ui.glyphs if self.frame: allowed = nomadnet.NomadNetworkApp.get_shared_instance().directory.is_known(bytes.fromhex(self.source_hash)) if allowed: self.frame.contents["footer"] = (self._build_footer(), None) else: warning = urwid.AttrMap( urwid.Padding(urwid.Text( "\n"+g["info"]+"\n\nYou cannot currently message this peer, since its identity keys are not known. " "The keys have been requested from the network and should arrive shortly, if available. " "Close this conversation and reopen it to try again.\n\n" "To query the network manually, select this conversation in the conversation list, " "press Ctrl-E, and use the query button.\n", align=urwid.CENTER, )), "msg_header_caution", ) self.frame.contents["footer"] = (warning, None) def toggle_focus_area(self): name = "" try: name = self.frame.get_focus_widgets()[0].name except Exception as e: pass if name == "messagelist": self.frame.focus_position = "footer" elif name == "minimal_editor" or name == "full_editor": self.frame.focus_position = "body" def keypress(self, size, key): if key == "tab": self.toggle_focus_area() elif key == "ctrl w": self.close() elif key == "ctrl u": self.conversation.purge_failed() self.conversation_changed(None) elif key == "ctrl t": self.toggle_editor() elif key == "ctrl x": self.clear_history_dialog() elif key == "ctrl g": nomadnet.NomadNetworkApp.get_shared_instance().ui.main_display.sub_displays.conversations_display.toggle_fullscreen() elif key == "ctrl o": self.sort_by_timestamp ^= True self.conversation_changed(None) elif key == "ctrl a": self.attach_file() elif key == "ctrl s": self.save_focused_attachments() else: return super(ConversationWidget, self).keypress(size, key) def _on_conversation_changed_from_callback(self, conversation): self.delegate._wake(lambda: self.conversation_changed(conversation)) def conversation_changed(self, conversation): if hasattr(self, "peer_info_widget"): self._update_peer_info() self.update_message_widgets(replace = True) def update_message_widgets(self, replace = False): self.message_widgets = [] added_hashes = [] needs_index = [] for message in self.conversation.messages: message_hash = message.get_hash() if not message_hash in added_hashes: added_hashes.append(message_hash) was_loaded = message.loaded try: message_widget = LXMessageWidget(message, theme=self.app.config["textui"]["theme"], conversation_widget=self) except Exception as e: RNS.log("Skipping message loading for "+str(message.file_path)+" due to error: "+str(e), RNS.LOG_WARNING) message.unload() continue self.message_widgets.append(message_widget) if not was_loaded and message.loaded: needs_index.append(message) message.unload() if needs_index: try: ConversationMessage.write_index( self.conversation.messages_path, needs_index) except Exception: pass if self.sort_by_timestamp: self.message_widgets.sort(key=lambda m: m.timestamp, reverse=False) else: self.message_widgets.sort(key=lambda m: m.sort_timestamp, reverse=False) from nomadnet.vendor.additional_urwid_widgets import IndicativeListBox self.messagelist = IndicativeListBox(self.message_widgets, position = len(self.message_widgets)-1) self.messagelist.name = "messagelist" if replace: self.frame.contents["body"] = (self.messagelist, None) nomadnet.NomadNetworkApp.get_shared_instance().ui.loop.draw_screen() def clear_editor(self): self.content_editor.set_edit_text("") self.title_editor.set_edit_text("") self.pending_attachments = [] self.frame.contents["footer"] = (self._build_footer(), None) def _collect_attachment_refs(self): g = self.app.ui.glyphs refs = [] sorted_messages = sorted(self.conversation.messages, key=lambda m: m.sort_timestamp, reverse=True) for conv_message in sorted_messages: if not conv_message.has_attachments(): continue cached_names = conv_message._cached_attachment_names or [] att_file_idx = 0 for atype, aname, *arest in cached_names: asize = arest[0] if arest else 0 glyph = g["file"] if atype == "file" else g[atype] label = glyph+" "+aname if asize > 0: label += " ("+_format_size(asize)+")" if atype == "file": refs.append((label, aname, conv_message, "file", att_file_idx)) att_file_idx += 1 else: refs.append((label, aname, conv_message, atype, 0)) return refs def save_focused_attachments(self): g = self.app.ui.glyphs self.dialog_active = True try: attachment_items = self._collect_attachment_refs() except Exception as e: RNS.log("Error collecting attachments: "+str(e), RNS.LOG_ERROR) attachment_items = [] save_dir = self.app.attachment_save_path if self.app.attachment_save_path else self.app.downloads_path def dismiss_dialog(sender): self.dialog_active = False self.conversation_changed(None) if not attachment_items: dialog = DialogLineBox( urwid.Pile([ urwid.Text("No attachments in this conversation.\n"), urwid.Columns([ (urwid.WEIGHT, 0.6, urwid.Text("")), (urwid.WEIGHT, 0.4, urwid.Button("OK", on_press=dismiss_dialog)), ]) ]), title="Attachments" ) dialog.delegate = self bottom = self.messagelist overlay = urwid.Overlay(dialog, bottom, align=urwid.CENTER, width=45, valign=urwid.MIDDLE, height=urwid.PACK, left=2, right=2) self.frame.contents["body"] = (overlay, self.frame.options()) self.frame.focus_position = "body" return checkboxes = [] for label, filename, conv_msg, field_type, field_index in attachment_items: cb = urwid.CheckBox(label, state=False) cb._attachment_filename = filename cb._conv_message = conv_msg cb._field_type = field_type cb._field_index = field_index checkboxes.append(cb) status_text = urwid.Text("") def do_save(sender): saved = [] errors = [] for cb in checkboxes: if cb.get_state(): try: src_path = cb._conv_message.get_attachment_file_path(cb._field_type, cb._field_index) if src_path and os.path.isfile(src_path): path = _copy_attachment_to_dest(cb._attachment_filename, src_path) saved.append(path) except Exception as e: errors.append(str(e)) if saved: lines = [g["check"]+" Copied "+str(len(saved))+" file(s) to "+save_dir+":"] for p in saved: lines.append(" "+os.path.basename(p)) if errors: lines.append(g["cross"]+" "+str(len(errors))+" failed") status_text.set_text("\n".join(lines)) elif errors: status_text.set_text(g["cross"]+" Failed: "+errors[0]) else: status_text.set_text("No files selected") dialog_widgets = list(checkboxes) dialog_widgets.append(urwid.Divider(g["divider1"])) dialog_widgets.append(urwid.Text("Copy to: "+save_dir)) dialog_widgets.append(status_text) dialog_widgets.append(urwid.Text("")) dialog_widgets.append(urwid.Columns([ (urwid.WEIGHT, 0.45, urwid.Button("Copy to Downloads", on_press=do_save)), (urwid.WEIGHT, 0.1, urwid.Text("")), (urwid.WEIGHT, 0.45, urwid.Button("Close", on_press=dismiss_dialog)), ])) dialog = DialogLineBox(urwid.ListBox(urwid.SimpleFocusListWalker(dialog_widgets)), title="Attachments") dialog.delegate = self bottom = self.messagelist overlay = urwid.Overlay(dialog, bottom, align=urwid.CENTER, width=("relative", 80), valign=urwid.MIDDLE, height=("relative", 80), left=2, right=2) self.frame.contents["body"] = (overlay, self.frame.options()) self.frame.focus_position = "body" def send_message(self): content = self.content_editor.get_edit_text() title = self.title_editor.get_edit_text() if not content == "": fields = None if self.pending_attachments: file_attachments = [] for file_path in self.pending_attachments: try: with open(file_path, "rb") as af: file_data = af.read() file_name = os.path.basename(file_path) file_attachments.append([file_name, file_data]) except Exception as e: RNS.log("Error reading attachment "+str(file_path)+": "+str(e), RNS.LOG_ERROR) if file_attachments: fields = {LXMF.FIELD_FILE_ATTACHMENTS: file_attachments} if self.app.compose_markdown: if not fields: fields = {} fields[LXMF.FIELD_RENDERER] = LXMF.RENDERER_MARKDOWN if self.conversation.send(content, title, fields=fields): self.clear_editor() def attach_file(self): self.dialog_active = True browser = FileBrowserDialog(self) bottom = self.messagelist overlay = urwid.Overlay(browser, bottom, align=urwid.CENTER, width=("relative", 90), valign=urwid.MIDDLE, height=("relative", 80), left=2, right=2) self.frame.contents["body"] = (overlay, self.frame.options()) self.frame.focus_position = "body" def file_browser_closed(self): self.dialog_active = False self.frame.contents["footer"] = (self._build_footer(), None) self.conversation_changed(None) def paper_message_saved(self, path): g = self.app.ui.glyphs def dismiss_dialog(sender): self.dialog_open = False self.conversation_changed(None) dialog = DialogLineBox( urwid.Pile([ urwid.Text("The paper message was saved to:\n\n"+str(path)+"\n", align=urwid.CENTER), urwid.Columns([ (urwid.WEIGHT, 0.6, urwid.Text("")), (urwid.WEIGHT, 0.4, urwid.Button("OK", on_press=dismiss_dialog)), ]) ]), title=g["papermsg"].replace(" ", "") ) dialog.delegate = self bottom = self.messagelist overlay = urwid.Overlay(dialog, bottom, align=urwid.CENTER, width=60, valign=urwid.MIDDLE, height=urwid.PACK, left=2, right=2) self.frame.contents["body"] = (overlay, self.frame.options()) self.frame.focus_position = "body" def print_paper_message_qr(self): content = self.content_editor.get_edit_text() title = self.title_editor.get_edit_text() if not content == "": if self.conversation.paper_output(content, title): self.clear_editor() else: self.paper_message_failed() def save_paper_message_qr(self): content = self.content_editor.get_edit_text() title = self.title_editor.get_edit_text() if not content == "": output_result = self.conversation.paper_output(content, title, mode="save_qr") if output_result != False: self.clear_editor() self.paper_message_saved(output_result) else: self.paper_message_failed() def save_paper_message_uri(self): content = self.content_editor.get_edit_text() title = self.title_editor.get_edit_text() if not content == "": output_result = self.conversation.paper_output(content, title, mode="save_uri") if output_result != False: self.clear_editor() self.paper_message_saved(output_result) else: self.paper_message_failed() def paper_message(self): def dismiss_dialog(sender): self.dialog_open = False self.conversation_changed(None) def print_qr(sender): dismiss_dialog(self) self.print_paper_message_qr() def save_qr(sender): dismiss_dialog(self) self.save_paper_message_qr() def save_uri(sender): dismiss_dialog(self) self.save_paper_message_uri() dialog = DialogLineBox( urwid.Pile([ urwid.Text( "Select the desired paper message output method.\nSaved files will be written to:\n\n"+str(self.app.downloads_path)+"\n", align=urwid.CENTER, ), urwid.Columns([ (urwid.WEIGHT, 0.5, urwid.Button("Print QR", on_press=print_qr)), (urwid.WEIGHT, 0.1, urwid.Text("")), (urwid.WEIGHT, 0.5, urwid.Button("Save QR", on_press=save_qr)), (urwid.WEIGHT, 0.1, urwid.Text("")), (urwid.WEIGHT, 0.5, urwid.Button("Save URI", on_press=save_uri)), (urwid.WEIGHT, 0.1, urwid.Text("")), (urwid.WEIGHT, 0.5, urwid.Button("Cancel", on_press=dismiss_dialog)) ]) ]), title="Create Paper Message" ) dialog.delegate = self bottom = self.messagelist overlay = urwid.Overlay(dialog, bottom, align=urwid.CENTER, width=60, valign=urwid.MIDDLE, height=urwid.PACK, left=2, right=2) self.frame.contents["body"] = (overlay, self.frame.options()) self.frame.focus_position = "body" def paper_message_failed(self): def dismiss_dialog(sender): self.dialog_open = False self.conversation_changed(None) dialog = DialogLineBox( urwid.Pile([ urwid.Text( "Could not output paper message,\ncheck your settings. See the log\nfile for any error messages.\n", align=urwid.CENTER, ), urwid.Columns([ (urwid.WEIGHT, 0.6, urwid.Text("")), (urwid.WEIGHT, 0.4, urwid.Button("OK", on_press=dismiss_dialog)), ]) ]), title="!" ) dialog.delegate = self bottom = self.messagelist overlay = urwid.Overlay(dialog, bottom, align=urwid.CENTER, width=34, valign=urwid.MIDDLE, height=urwid.PACK, left=2, right=2) self.frame.contents["body"] = (overlay, self.frame.options()) self.frame.focus_position = "body" def close(self): self.delegate.close_conversation(self) class LXMessageWidget(urwid.WidgetWrap): mdc = MarkdownToMicron(max_width=80, syntax_highlighter=SyntaxHighlighter(), url_scope=None) def __init__(self, message, theme=THEME_DARK, conversation_widget=None): app = nomadnet.NomadNetworkApp.get_shared_instance() g = app.ui.glyphs self._conversation_widget = conversation_widget self.timestamp = message.get_timestamp() self.sort_timestamp = message.sort_timestamp self.transfer_done = False self._live_lxm = None msg_hash = message.get_hash() msg_state = message.get_state() msg_source_hash = message._cached_source_hash msg_method = message._cached_method time_format = app.time_format message_time = datetime.fromtimestamp(self.timestamp) renderer = message.content_renderer() encryption_string = "" if message.get_transport_encrypted(): encryption_string = " "+g["encrypted"] else: encryption_string = " "+g["plaintext"] title_string = relative_time(self.timestamp)+" | "+message_time.strftime(time_format)+encryption_string is_outbound = False if msg_source_hash is None: header_style = "msg_header_failed" title_string = g["warning"]+" "+title_string elif app.lxmf_destination.hash == msg_source_hash: is_outbound = True if msg_state == LXMF.LXMessage.DELIVERED: header_style = "msg_header_delivered" title_string = g["check"]+" "+g["arrow_r"]+" "+title_string elif msg_state == LXMF.LXMessage.FAILED: header_style = "msg_header_failed" title_string = g["cross"]+" "+g["arrow_r"]+" "+title_string elif msg_state == LXMF.LXMessage.REJECTED: header_style = "msg_header_failed" title_string = g["cross"]+" "+g["arrow_r"]+" Rejected "+title_string elif msg_method == LXMF.LXMessage.PROPAGATED and msg_state == LXMF.LXMessage.SENT: header_style = "msg_header_propagated" title_string = g["sent"]+" "+g["arrow_r"]+" "+title_string elif msg_method == LXMF.LXMessage.PAPER and msg_state == LXMF.LXMessage.PAPER: header_style = "msg_header_propagated" title_string = g["papermsg"]+" "+g["arrow_r"]+" "+title_string elif msg_state == LXMF.LXMessage.SENT: header_style = "msg_header_sent" title_string = g["sent"]+" "+g["arrow_r"]+" "+title_string else: header_style = "msg_header_sent" title_string = g["arrow_r"]+" "+title_string else: if message.signature_validated(): header_style = "msg_header_ok" title_string = g["check"]+" "+g["arrow_l"]+" "+title_string else: header_style = "msg_header_caution" title_string = g["warning"]+" "+g["arrow_l"]+" "+message.get_signature_description() + "\n " + title_string if message.get_title() != "": title_string += " | " + message.get_title() inbound_untrusted = False if not is_outbound and msg_source_hash is not None: try: sender_trust = app.directory.trust_level(msg_source_hash) if sender_trust in (DirectoryEntry.UNTRUSTED, DirectoryEntry.WARNING, DirectoryEntry.UNKNOWN): inbound_untrusted = True except Exception: inbound_untrusted = False has_attachments = message.has_attachments() cached_names = message._cached_attachment_names or [] if has_attachments and cached_names: attachment_strings = [] for atype, aname, *arest in cached_names: attachment_strings.append(g[atype if atype != "file" else "file"]+" "+aname) title_string += " | " + " ".join(attachment_strings) content_text = message.get_content() if content_text: copy_glyph = g.get("copy", "[C]") check_glyph = g.get("check", "v").center(len(copy_glyph)) copy_icon = ClickableIcon(copy_glyph) conv_widget = self._conversation_widget def on_copy_click(icon=copy_icon, content=content_text, cg=copy_glyph, kg=check_glyph, cw=conv_widget): osc52_copy(content) icon.set_text(kg) def _restore(loop, user_data): icon.set_text(cg) try: app.ui.loop.set_alarm_in(2.0, _restore) except Exception: icon.set_text(cg) if cw is not None and cw.frame is not None: def _refocus(loop, user_data): try: cw.frame.focus_position = "footer" except Exception: pass try: app.ui.loop.set_alarm_in(0, _refocus) except Exception: pass copy_icon._on_click = on_copy_click copy_width = len(copy_glyph) + 2 title_row = urwid.Columns([ ("weight", 1, urwid.Text(title_string)), (copy_width, urwid.Padding(copy_icon, left=1, right=1)), ]) title = urwid.AttrMap(title_row, header_style) else: title = urwid.AttrMap(urwid.Text(title_string), header_style) self.progress_widget = urwid.Text("") self.progress_attr = urwid.AttrMap(self.progress_widget, "progress_full") content_lines = content_text.split("\n") markdown = renderer == LXMF.RENDERER_MARKDOWN default_fg = "bbb" if theme == THEME_DARK else "444" if markdown: formatted = self.mdc.format_block(content_text) message_body = strip_non_formatting_tags(formatted) rendered = markup_to_attrmaps(strip_modifiers(message_body), url_delegate=None, fg_color=default_fg, bg_color=None) content_pile = urwid.Padding(urwid.Pile(rendered), left=2, right=2) else: indented = "\n".join(" "+line for line in content_lines) pile_widgets = [title] if is_outbound and msg_state is not None and msg_state < LXMF.LXMessage.SENT and msg_hash is not None: try: for pending in app.message_router.pending_outbound: if pending.hash == msg_hash: if pending.representation == LXMF.LXMessage.RESOURCE: self._live_lxm = pending break except Exception: pass if self._live_lxm is not None: pct = int(self._live_lxm.progress * 100) bar_width = 20 filled = int(bar_width * self._live_lxm.progress) if app.ui.colormode >= 256: bar = "\u2588" * filled + "\u2591" * (bar_width - filled) else: bar = "#" * filled + "-" * (bar_width - filled) self.progress_widget.set_text(" ["+bar+"] "+str(pct)+"%") pile_widgets.append(self.progress_attr) self._start_progress_poll() if markdown: pile_widgets.append(content_pile) else: pile_widgets.append(urwid.Text(indented)) if has_attachments and cached_names: if inbound_untrusted: pile_widgets.append(urwid.AttrMap( urwid.Text(" "+g["warning"]+" This attachment came from a peer that's untrusted. Be careful when opening it."), "list_untrusted", )) att_file_idx = 0 for atype, aname, *arest in cached_names: glyph = g["file"] if atype == "file" else g[atype] asize = arest[0] if arest else 0 label = " "+glyph+" "+aname if asize > 0: label += " ("+_format_size(asize)+")" if atype == "file": pile_widgets.append(ClickableAttachment(label, aname, message, "file", att_file_idx)) att_file_idx += 1 else: pile_widgets.append(ClickableAttachment(label, aname, message, atype)) pile_widgets.append(urwid.Text("")) super().__init__(urwid.Pile(pile_widgets)) def _start_progress_poll(self): try: loop = nomadnet.NomadNetworkApp.get_shared_instance().ui.loop if loop: loop.set_alarm_in(0.3, self._poll_progress) except Exception: pass def _poll_progress(self, loop=None, user_data=None): if self.transfer_done: return if self._live_lxm is None: self.transfer_done = True return app = nomadnet.NomadNetworkApp.get_shared_instance() g = app.ui.glyphs progress = self._live_lxm.progress state = self._live_lxm.state pct = int(progress * 100) if state == LXMF.LXMessage.FAILED: self.progress_widget.set_text(" "+g["cross"]+" Transfer failed") self.transfer_done = True self._live_lxm = None elif state == LXMF.LXMessage.REJECTED: self.progress_widget.set_text(" "+g["cross"]+" Rejected: too large or not accepted") self.transfer_done = True self._live_lxm = None elif state >= LXMF.LXMessage.SENT: self.progress_widget.set_text("") self.transfer_done = True self._live_lxm = None else: bar_width = 20 filled = int(bar_width * progress) if app.ui.colormode >= 256: bar = "\u2588" * filled + "\u2591" * (bar_width - filled) else: bar = "#" * filled + "-" * (bar_width - filled) self.progress_widget.set_text(" ["+bar+"] "+str(pct)+"%") if not self.transfer_done: try: ui_loop = app.ui.loop if ui_loop: ui_loop.set_alarm_in(0.3, self._poll_progress) ui_loop.draw_screen() except Exception: pass class ClickableAttachment(urwid.Text): def __init__(self, label, filename, conv_message, field_type, field_index=0): self.filename = filename self.conv_message = conv_message self.field_type = field_type self.field_index = field_index self.saved = False super().__init__(label) def mouse_event(self, size, event, button, x, y, focus): if button == 1 and urwid.util.is_mouse_press(event): self._save() return True return False def _save(self): if self.saved: return app = nomadnet.NomadNetworkApp.get_shared_instance() g = app.ui.glyphs try: src_path = self.conv_message.get_attachment_file_path(self.field_type, self.field_index) if src_path and os.path.isfile(src_path): save_path = _copy_attachment_to_dest(self.filename, src_path) else: if self.field_type == "file": attachments = self.conv_message.get_file_attachments() if self.field_index < len(attachments): att = attachments[self.field_index] if isinstance(att, list) and len(att) >= 2: data = att[1] if isinstance(att[1], bytes) else b"" else: data = b"" else: data = b"" elif self.field_type == "image": data = self.conv_message.get_image() data = data if isinstance(data, bytes) else b"" elif self.field_type == "audio": data = self.conv_message.get_audio() data = data if isinstance(data, bytes) else b"" else: data = b"" self.conv_message.unload() if not data: return save_path = _save_attachment_to_disk(self.filename, data) self.saved = True self.set_text(" "+g["check"]+" Copied to: "+save_path) except Exception as e: RNS.log("Error saving attachment: "+str(e), RNS.LOG_ERROR) self.set_text(" "+g["cross"]+" Save failed: "+str(e)) def _resolve_attachment_save_path(filename): app = nomadnet.NomadNetworkApp.get_shared_instance() save_dir = app.attachment_save_path if app.attachment_save_path else app.downloads_path if not os.path.isdir(save_dir): os.makedirs(save_dir) safe_name = ConversationMessage.safe_attachment_name(filename) base_dir = os.path.realpath(save_dir) + os.sep candidate = os.path.realpath(os.path.join(save_dir, safe_name)) if not (candidate + os.sep).startswith(base_dir): raise OSError(13, os.strerror(13)) counter = 0 base, ext = os.path.splitext(safe_name) while os.path.isfile(candidate): counter += 1 candidate = os.path.realpath(os.path.join(save_dir, base+"_"+str(counter)+ext)) if not (candidate + os.sep).startswith(base_dir): raise OSError(13, os.strerror(13)) return candidate def _copy_attachment_to_dest(filename, src_path): save_path = _resolve_attachment_save_path(filename) shutil.copy2(src_path, save_path) return save_path def _save_attachment_to_disk(filename, data): save_path = _resolve_attachment_save_path(filename) with open(save_path, "wb") as f: f.write(data) return save_path class FileBrowserEntry(urwid.WidgetWrap): signals = ["click"] def __init__(self, name, full_path, is_dir=False, is_parent=False, selected=False): self.full_path = full_path self.name = name self.is_dir = is_dir self.is_parent = is_parent self.selected = selected g = nomadnet.NomadNetworkApp.get_shared_instance().ui.glyphs if is_parent: display = g["arrow_l"]+" .." elif is_dir: display = g["arrow_r"]+" "+name+"/" elif selected: display = g["check"]+" "+name else: display = " "+name self.text_widget = urwid.SelectableIcon(display, 0) if is_dir or is_parent: style = "list_trusted" focus_style = "list_focus" elif selected: style = "list_trusted" focus_style = "list_focus_trusted" else: style = "list_unknown" focus_style = "list_focus" display_widget = urwid.AttrMap(self.text_widget, style, focus_style) super().__init__(display_widget) def keypress(self, size, key): if key == "enter": self._emit("click") else: return key def mouse_event(self, size, event, button, x, y, focus): if button == 1 and urwid.util.is_mouse_press(event): self._emit("click") return True return False class FileBrowserDialog(urwid.WidgetWrap): def __init__(self, delegate): self.delegate = delegate app = nomadnet.NomadNetworkApp.get_shared_instance() self.g = app.ui.glyphs self.current_path = os.path.expanduser("~") self.path_label = urwid.Text("") self.status_label = urwid.Text("") self.file_walker = urwid.SimpleFocusListWalker([]) self.file_listbox = urwid.ListBox(self.file_walker) self.button_columns = urwid.Columns([ (urwid.WEIGHT, 0.45, urwid.Button("Done", on_press=self._dismiss)), (urwid.WEIGHT, 0.1, urwid.Text("")), (urwid.WEIGHT, 0.45, urwid.Button("Cancel", on_press=self._cancel)), ]) header_pile = urwid.Pile([ self.path_label, self.status_label, urwid.Divider(self.g["divider1"]), ]) footer_pile = urwid.Pile([ urwid.Divider(self.g["divider1"]), self.button_columns, ]) self._populate() self.browser_frame = urwid.Frame( self.file_listbox, header=header_pile, footer=footer_pile, ) linebox = urwid.LineBox(self.browser_frame, title="Attach File") super().__init__(linebox) def _update_status(self): pending = self.delegate.pending_attachments if pending: names = [os.path.basename(p) for p in pending] self.status_label.set_text(" "+self.g["file"]+" "+str(len(pending))+" selected: "+", ".join(names)) else: self.status_label.set_text(" No files selected") def _populate(self): self.path_label.set_text(" "+self.current_path) self._update_status() focus_pos = None try: focus_pos = self.file_listbox.focus_position except Exception: pass entries = [] parent = os.path.dirname(self.current_path) if parent != self.current_path: entry = FileBrowserEntry("..", parent, is_parent=True) urwid.connect_signal(entry, "click", self._entry_clicked, entry) entries.append(entry) try: items = sorted(os.listdir(self.current_path)) except PermissionError: entries.append(urwid.Text(("error_text", " Permission denied"))) self.file_walker[:] = entries return dirs = [] files = [] for item in items: if item.startswith("."): continue full = os.path.join(self.current_path, item) if os.path.isdir(full): dirs.append((item, full)) elif os.path.isfile(full): files.append((item, full)) for name, full in dirs: entry = FileBrowserEntry(name, full, is_dir=True) urwid.connect_signal(entry, "click", self._entry_clicked, entry) entries.append(entry) for name, full in files: is_selected = full in self.delegate.pending_attachments entry = FileBrowserEntry(name, full, selected=is_selected) urwid.connect_signal(entry, "click", self._entry_clicked, entry) entries.append(entry) if not dirs and not files: entries.append(urwid.Text(("inactive_text", " (empty)"))) self.file_walker[:] = entries if focus_pos is not None and focus_pos < len(entries): self.file_listbox.set_focus(focus_pos) elif entries: self.file_listbox.set_focus(0) def _entry_clicked(self, entry_widget, user_data=None): entry = user_data if user_data else entry_widget if entry.is_dir or entry.is_parent: self.current_path = entry.full_path self._populate() else: if entry.full_path in self.delegate.pending_attachments: self.delegate.pending_attachments.remove(entry.full_path) else: self.delegate.pending_attachments.append(entry.full_path) self.delegate.frame.contents["footer"] = (self.delegate._build_footer(), None) self._populate() def _dismiss(self, sender): self.delegate.file_browser_closed() def _cancel(self, sender): self.delegate.pending_attachments.clear() self.delegate.frame.contents["footer"] = (self.delegate._build_footer(), None) self.delegate.file_browser_closed() def keypress(self, size, key): if key == "esc": self.delegate.file_browser_closed() return result = super().keypress(size, key) if result == "down" and self.browser_frame.focus_position == "body": self.browser_frame.focus_position = "footer" return elif result == "up" and self.browser_frame.focus_position == "footer": self.browser_frame.focus_position = "body" return return result class SyncProgressBar(urwid.ProgressBar): def get_text(self): status = nomadnet.NomadNetworkApp.get_shared_instance().get_sync_status() show_percent = nomadnet.NomadNetworkApp.get_shared_instance().sync_status_show_percent() if show_percent: return status+" "+super().get_text() else: return status