beba425868
macro_manager: - All macro reads/writes guarded by a lock; macros.json written atomically (temp + fsync + os.replace). Commands are copied under the lock and executed outside it so wait/keyboard steps don't block other threads. - last_used disk writes debounced (>=5s) instead of a full rewrite per press. - Validate key/hotkey names against pyautogui.KEYBOARD_KEYS and clamp wait ms; invalid input is skipped/logged, not blindly executed. web_server: - WebSocket receive loop always disconnects (fixes dead-socket leak); broadcast prunes failed sockets. - POST /api/execute offloaded to a thread executor so a long macro no longer blocks the event loop. - Image upload hardened: 5MB cap (413), real-image validation via Pillow, server-generated name under macro_images/, returns a relative path only, cleans up on failure. gui: - Execute macros off the Qt main thread (signal back to status bar); debounce resizeEvent and only rebuild when the column count changes. - Relay password stored in the OS keyring when available (plaintext JSON blanked; graceful fallback if no backend); delete dialogs name the macro; honor the minimize-to-tray setting; surface save failures. - Add media/system keys (volume, play/pause, track nav) to the macro editor. - THEME gains shared danger/accent-hover colors; drop dead imports. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
172 lines
5.7 KiB
Python
172 lines
5.7 KiB
Python
# Settings Manager for MacroPad Server
|
|
|
|
import os
|
|
import json
|
|
import secrets
|
|
from typing import Any, Optional
|
|
|
|
# Optional OS keyring for storing the relay password at rest. If the library or
|
|
# a working backend is unavailable we fall back to the JSON file (now chmod 0600).
|
|
try:
|
|
import keyring as _keyring
|
|
except Exception: # ImportError or any backend import-time failure
|
|
_keyring = None
|
|
|
|
KEYRING_SERVICE = "MacroPad"
|
|
KEYRING_USER = "relay"
|
|
|
|
|
|
DEFAULT_SETTINGS = {
|
|
"relay": {
|
|
"enabled": False,
|
|
"server_url": "wss://relay.macropad.example.com",
|
|
"session_id": None,
|
|
"password": ""
|
|
},
|
|
"web": {
|
|
"auth_token": "",
|
|
"allow_lan": True
|
|
},
|
|
"minimize_to_tray": True
|
|
}
|
|
|
|
|
|
class SettingsManager:
|
|
"""Manages application settings with JSON persistence."""
|
|
|
|
def __init__(self, settings_file: str):
|
|
self.settings_file = settings_file
|
|
self.settings = {}
|
|
self.load()
|
|
|
|
def load(self):
|
|
"""Load settings from file."""
|
|
if os.path.exists(self.settings_file):
|
|
try:
|
|
with open(self.settings_file, 'r', encoding='utf-8') as f:
|
|
self.settings = json.load(f)
|
|
except (json.JSONDecodeError, IOError):
|
|
self.settings = {}
|
|
|
|
# Merge with defaults to ensure all keys exist
|
|
self.settings = self._merge_defaults(DEFAULT_SETTINGS, self.settings)
|
|
|
|
# Ensure a web auth token exists (this is the security boundary for API access)
|
|
if not self.get('web.auth_token'):
|
|
token = secrets.token_urlsafe(32)
|
|
self.set('web.auth_token', token)
|
|
self.save()
|
|
|
|
def save(self):
|
|
"""Save settings to file."""
|
|
try:
|
|
os.makedirs(os.path.dirname(self.settings_file), exist_ok=True)
|
|
with open(self.settings_file, 'w', encoding='utf-8') as f:
|
|
json.dump(self.settings, f, indent=2)
|
|
# Best-effort restrict permissions (protects stored relay password + token)
|
|
try:
|
|
os.chmod(self.settings_file, 0o600)
|
|
except OSError:
|
|
pass
|
|
return True
|
|
except IOError:
|
|
return False
|
|
|
|
def _merge_defaults(self, defaults: dict, current: dict) -> dict:
|
|
"""Merge current settings with defaults, keeping current values."""
|
|
result = defaults.copy()
|
|
for key, value in current.items():
|
|
if key in result:
|
|
if isinstance(result[key], dict) and isinstance(value, dict):
|
|
result[key] = self._merge_defaults(result[key], value)
|
|
else:
|
|
result[key] = value
|
|
else:
|
|
result[key] = value
|
|
return result
|
|
|
|
def get(self, key: str, default: Any = None) -> Any:
|
|
"""Get a setting value by key (supports dot notation)."""
|
|
keys = key.split('.')
|
|
value = self.settings
|
|
try:
|
|
for k in keys:
|
|
value = value[k]
|
|
return value
|
|
except (KeyError, TypeError):
|
|
return default
|
|
|
|
def set(self, key: str, value: Any):
|
|
"""Set a setting value by key (supports dot notation)."""
|
|
keys = key.split('.')
|
|
target = self.settings
|
|
for k in keys[:-1]:
|
|
if k not in target:
|
|
target[k] = {}
|
|
target = target[k]
|
|
target[keys[-1]] = value
|
|
|
|
def get_relay_enabled(self) -> bool:
|
|
"""Check if relay server is enabled."""
|
|
return self.get('relay.enabled', False)
|
|
|
|
def get_relay_url(self) -> str:
|
|
"""Get the relay server URL."""
|
|
return self.get('relay.server_url', '')
|
|
|
|
def get_relay_session_id(self) -> Optional[str]:
|
|
"""Get the stored relay session ID."""
|
|
return self.get('relay.session_id')
|
|
|
|
def get_relay_password(self) -> str:
|
|
"""Get the relay password, preferring the OS keyring when available."""
|
|
if _keyring is not None:
|
|
try:
|
|
value = _keyring.get_password(KEYRING_SERVICE, KEYRING_USER)
|
|
if value:
|
|
return value
|
|
except Exception:
|
|
# Keyring backend missing/broken: fall back to JSON.
|
|
pass
|
|
return self.get('relay.password', '')
|
|
|
|
def set_relay_password(self, password: str) -> bool:
|
|
"""Store the relay password, preferring the OS keyring when available.
|
|
|
|
When keyring works, the plaintext password is kept out of the JSON file.
|
|
Never crashes if the keyring backend is missing; falls back to JSON.
|
|
Returns True on successful persistence.
|
|
"""
|
|
password = password or ""
|
|
stored_in_keyring = False
|
|
if _keyring is not None:
|
|
try:
|
|
_keyring.set_password(KEYRING_SERVICE, KEYRING_USER, password)
|
|
stored_in_keyring = True
|
|
except Exception:
|
|
stored_in_keyring = False
|
|
|
|
if stored_in_keyring:
|
|
# Don't persist plaintext in the JSON when the keyring holds it.
|
|
self.set('relay.password', "")
|
|
else:
|
|
self.set('relay.password', password)
|
|
return self.save()
|
|
|
|
def set_relay_session_id(self, session_id: str) -> bool:
|
|
"""Store the relay session ID. Returns True on successful save."""
|
|
self.set('relay.session_id', session_id)
|
|
return self.save()
|
|
|
|
def get_web_auth_token(self) -> str:
|
|
"""Get the web API auth token."""
|
|
return self.get('web.auth_token', '')
|
|
|
|
def get_web_allow_lan(self) -> bool:
|
|
"""Check if LAN (0.0.0.0) binding is allowed."""
|
|
return self.get('web.allow_lan', True)
|
|
|
|
def get_minimize_to_tray(self) -> bool:
|
|
"""Check if minimize to tray is enabled."""
|
|
return self.get('minimize_to_tray', True)
|