rt/NomadNet
1
0
Fork 0
mirror of https://github.com/markqvist/NomadNet.git synced 2026-10-25 18:40:13 -04:00
Code Issues Projects Releases Packages Wiki Activity Actions

Atomic writes for config fields, don't overwrite intro_time, Node editor allow fixes

Browse source
This commit is contained in:
zenith 2026-08-03 17:27:47 -04:00
parent 0d3b519a07
commit 5d0ffce51c
5 changed files with 143 additions and 46 deletions
Download patch file Download diff file Expand all files Collapse all files

3
nomadnet/InterfaceProfiles.py
View file

@ -3,6 +3,7 @@ import threading
import RNS
import RNS.vendor.umsgpack as msgpack
from nomadnet.util import write_config_atomic
class InterfaceProfiles:
@ -178,7 +179,7 @@ class InterfaceProfiles:
interfaces[name]["interface_enabled"] = value
if "enabled" in interfaces[name]:
interfaces[name]["enabled"] = value
self.app.rns.config.write()
write_config_atomic(self.app.rns.config)
return True
except Exception as e:
RNS.log("Could not apply interface profile: "+str(e), RNS.LOG_ERROR)

56
nomadnet/ui/textui/Config.py
View file

@ -4,6 +4,7 @@ import urwid
import platform
from RNS.vendor.configobj import ConfigObj
from nomadnet.util import write_config_atomic
from nomadnet.vendor.additional_urwid_widgets.FormWidgets import *
CONFIG_SECTION_GLYPHS = {
@ -208,12 +209,11 @@ CONFIG_SECTIONS = [
"fields": [
{
"config_key": "intro_time",
"label": "Show intro: ",
"type": "checkbox",
"default": True,
"on_value": "1.0",
"off_value": "0",
"help": "Show the intro screen at startup",
"label": "Intro time: ",
"type": "edit",
"default": "1",
"validation": ["float"],
"help": "Seconds to show the intro screen at startup (0 to disable)",
},
{
"config_key": "intro_text",
@ -976,6 +976,21 @@ class ConfigSectionView(urwid.WidgetWrap):
all_valid = False
return all_valid
def _widget_write_value(self, field, widget):
if field["type"] == "checkbox":
return field.get("on_value", "yes") if widget.get_state() else field.get("off_value", "no")
return widget.get_value()
def _default_write_value(self, field):
default = field.get("default")
if field["type"] == "checkbox":
return field.get("on_value", "yes") if bool(default) else field.get("off_value", "no")
if default is None:
return ""
if isinstance(default, (list, tuple)):
return list(default)
return str(default)
def on_save(self, button):
if not self.validate_all():
self.show_message("Some fields have invalid values.\nCorrect the marked fields and try again.", title="Not saved", back_to_list=False)
@ -986,7 +1001,8 @@ class ConfigSectionView(urwid.WidgetWrap):
except Exception:
pass
if self.section_key not in self.file_config:
section_existed = self.section_key in self.file_config
if not section_existed:
self.file_config[self.section_key] = {}
section_config = self.file_config[self.section_key]
@ -994,26 +1010,20 @@ class ConfigSectionView(urwid.WidgetWrap):
for config_key, entry in self.fields.items():
field = entry["field"]
widget = entry["widget"]
present = config_key in section_config
value = self._widget_write_value(field, widget)
if field["type"] == "checkbox":
on_value = field.get("on_value", "yes")
off_value = field.get("off_value", "no")
section_config[config_key] = on_value if widget.get_state() else off_value
elif field["type"] == "list":
value = widget.get_value()
if value:
section_config[config_key] = value
elif config_key in section_config:
del section_config[config_key]
else:
value = widget.get_value()
if value != "":
section_config[config_key] = value
elif config_key in section_config:
if not value:
if present:
del section_config[config_key]
elif present or value != self._default_write_value(field):
section_config[config_key] = value
if not section_existed and not section_config:
del self.file_config[self.section_key]
try:
self.file_config.write()
write_config_atomic(self.file_config)
if self.section.get("configfile") == "reticulum":
try:
self.parent.app.rns.config.reload()

82
nomadnet/ui/textui/Editor.py
View file

@ -912,36 +912,88 @@ class PageEditorDisplay():
except Exception as e:
self.set_status("delete failed: %s" % e)
def _read_allowed(self, path):
ap = path + ".allowed"
if not os.path.isfile(ap):
return ""
try:
with open(ap, encoding="utf-8") as f:
return "\n".join(ln.strip() for ln in f.read().splitlines() if ln.strip())
except Exception:
return ""
def _walk_pages(self, path):
for root, dirs, files in os.walk(path):
dirs[:] = [d for d in dirs if not d.startswith(".") and d != "__pycache__"]
for fn in files:
if fn.endswith(".allowed"):
continue
yield os.path.join(root, fn)
def _folder_permissions(self, path):
bodies = {}
total = 0
for page in self._walk_pages(path):
total += 1
body = self._read_allowed(page)
bodies[body] = bodies.get(body, 0) + 1
restricted = sum(count for body, count in bodies.items() if body)
distinct = [body for body in bodies if body]
common = distinct[0] if len(bodies) == 1 and distinct else None
return common, restricted, total
def permissions_dialog(self):
w = self._selected_tree()
if not w or not getattr(w, "path", None):
self.set_status("select a file/folder for permissions"); return
path = w.path
is_dir = os.path.isdir(path)
content = ""
if not is_dir and os.path.isfile(path + ".allowed"):
try: content = open(path + ".allowed", encoding="utf-8").read()
except Exception: content = ""
note = ("Allowed identity hashes per line. Empty = public.\n"
+ ("Applies to every page in this folder." if is_dir else "Applys to this page."))
if is_dir:
common, restricted, total = self._folder_permissions(path)
content = common or ""
if not total:
state = "This folder contains no pages."
elif not restricted:
state = "All %d pages here are public." % total
elif common is not None:
state = "All %d pages here share these entries." % total
else:
state = "Entries differ across pages (%d of %d restricted)." % (restricted, total)
note = ("Allowed identity hashes per line.\n"
+ state + "\nSaving applies these entries to every page in this folder and its subfolders.")
else:
content = self._read_allowed(path)
note = "Allowed identity hashes per line. Empty = public.\nApplies to this page."
title = "Permissions: " + os.path.basename(path) + ("/" if is_dir else "")
self.show_dialog(PermissionsDialog(title, note, content,
lambda txt: self._save_permissions(path, is_dir, txt), self.close_dialog))
def _save_permissions(self, path, is_dir, text):
self.close_dialog()
body = "\n".join(ln.strip() for ln in text.splitlines() if ln.strip())
if is_dir and not body:
restricted = self._folder_permissions(path)[1]
if restricted:
self.show_dialog(ConfirmDialog("Make public",
"This removes the access restrictions from %d page(s)\nunder %s, making them public.\n\nContinue?"
% (restricted, os.path.basename(path)),
[("Make public", lambda: self._apply_permissions(path, is_dir, body)), ("Cancel", self.close_dialog)],
on_cancel=self.close_dialog))
return
self._apply_permissions(path, is_dir, body)
def _apply_permissions(self, path, is_dir, body):
self.close_dialog()
try:
if is_dir:
count = 0
for root, dirs, files in os.walk(path):
dirs[:] = [d for d in dirs if not d.startswith(".") and d != "__pycache__"]
for fn in files:
if fn.endswith(".allowed"):
continue
self._write_allowed(os.path.join(root, fn), body)
count += 1
self.set_status("permissions applied to %d files" % count)
for page in self._walk_pages(path):
self._write_allowed(page, body)
count += 1
self.set_status(("permissions applied to %d files" if body else "permissions cleared on %d files") % count)
else:
self._write_allowed(path, body)
self.set_status("permissions saved" if body else "permissions cleared (public)")

15
nomadnet/ui/textui/Interfaces.py
View file

@ -8,6 +8,7 @@ from nomadnet.vendor.additional_urwid_widgets.FormWidgets import *
from nomadnet.vendor.AsciiChart import AsciiChart
from .ReadlineEdit import ReadlineEdit
from .Helpers import ClickableIcon
from nomadnet.util import write_config_atomic
### GYLPHS ###
INTERFACE_GLYPHS = {
@ -2066,7 +2067,7 @@ class AddInterfaceView(urwid.WidgetWrap):
try:
interfaces = self.parent.app.rns.config['interfaces']
interfaces[name] = interface_config
self.parent.app.rns.config.write()
write_config_atomic(self.parent.app.rns.config)
self._apply_profile_membership(name)
display_type = custom_type if self.iface_type == "CustomInterface" else self.iface_type
@ -2323,7 +2324,7 @@ class EditInterfaceView(AddInterfaceView):
else:
interfaces[self.iface_name] = updated_config
self.parent.app.rns.config.write()
write_config_atomic(self.parent.app.rns.config)
self._apply_profile_membership(new_name)
display_type = interface_type
@ -2798,7 +2799,7 @@ class ShowInterface(urwid.WidgetWrap):
try:
interfaces = self.parent.app.rns.config['interfaces']
interfaces[self.iface_name] = self.interface_config
self.parent.app.rns.config.write()
write_config_atomic(self.parent.app.rns.config)
self.update_status_display()
@ -3742,7 +3743,7 @@ class InterfaceDisplay:
try:
for name, conf in parsed.items():
interfaces[name] = conf
self.app.rns.config.write()
write_config_atomic(self.app.rns.config)
except Exception as e:
self.paste_status.set_text(("error", " Failed: " + str(e)))
return
@ -3765,7 +3766,7 @@ class InterfaceDisplay:
try:
if interface_name in self.app.rns.config['interfaces']:
del self.app.rns.config['interfaces'][interface_name]
self.app.rns.config.write()
write_config_atomic(self.app.rns.config)
try: self.app.interface_profiles.remove_interface(interface_name)
except Exception: pass
@ -3959,7 +3960,7 @@ class InterfaceDisplay:
if "enabled" in iface:
iface["enabled"] = value
try:
self.app.rns.config.write()
write_config_atomic(self.app.rns.config)
except Exception:
return
self.mark_restart_pending()
@ -4038,7 +4039,7 @@ class InterfaceDisplay:
changed += 1
try:
self.app.rns.config.write()
write_config_atomic(self.app.rns.config)
except Exception as e:
self._bulk_notice(f"Error applying bulk action: {str(e)}", title="Error")
return

33
nomadnet/util.py
View file

@ -1,6 +1,39 @@
import os
import re
import unicodedata
def write_config_atomic(config):
path = getattr(config, "filename", None)
if not path:
config.write()
return
directory = os.path.dirname(path) or "."
tmp = os.path.join(directory, "."+os.path.basename(path)+".tmp")
try:
try:
existing_mode = os.stat(path).st_mode
except OSError:
existing_mode = None
with open(tmp, "wb") as tmp_file:
config.write(tmp_file)
tmp_file.flush()
try: os.fsync(tmp_file.fileno())
except Exception: pass
if existing_mode is not None:
try: os.chmod(tmp, existing_mode)
except Exception: pass
os.replace(tmp, path)
except Exception:
try: os.remove(tmp)
except Exception: pass
raise
invalid_rendering = ["🕵️", "☝"]
# Unicode blocks to strip (symbols, dingbats, emoji ranges)