shadowdao
1f26427328
The local FastAPI server previously exposed macro create/execute over the
network with no authentication while bound to 0.0.0.0, and macro `app`
commands ran via subprocess with shell=True — an unauthenticated RCE for
anyone on the LAN (and, via the relay, the internet).
- Mandatory per-install auth token (secrets.token_urlsafe) required on all
/api/* routes and the /ws handshake; timing-safe hmac.compare_digest.
PWA shell + /static assets remain public to bootstrap.
- Token auto-generated and persisted in settings.json (now chmod 0600).
- Bind host is configurable (web.allow_lan); token is the security boundary.
- macro `app` launch no longer uses shell=True — commands are tokenized with
shlex and exec'd directly, killing ;/&&/|/$() shell chaining.
- Relay client injects the local token and only forwards /api/* paths, and no
longer forwards attacker-supplied headers to the local server.
- Fixed path traversal in /api/image/{path} (realpath-confined to macro_images).
- GUI embeds the token in the LAN URL/QR; PWA reads it and sends it on API,
WebSocket, and image requests.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
374 lines
12 KiB
Python
374 lines
12 KiB
Python
# Macro management and execution with command sequence support
|
|
|
|
import copy
|
|
import json
|
|
import os
|
|
import shlex
|
|
import uuid
|
|
import pyautogui
|
|
import subprocess
|
|
import time
|
|
from PIL import Image
|
|
from typing import Optional
|
|
|
|
|
|
class MacroManager:
|
|
"""Manages macro storage, migration, and execution with command sequences."""
|
|
|
|
def __init__(self, data_file: str, images_dir: str, app_dir: str):
|
|
self.data_file = data_file
|
|
self.images_dir = images_dir
|
|
self.app_dir = app_dir
|
|
self.macros = {}
|
|
self.load_macros()
|
|
|
|
def load_macros(self):
|
|
"""Load macros from JSON file and migrate if needed."""
|
|
try:
|
|
if os.path.exists(self.data_file):
|
|
with open(self.data_file, "r") as file:
|
|
self.macros = json.load(file)
|
|
|
|
# Migrate old format macros
|
|
migrated = False
|
|
for macro_id, macro in list(self.macros.items()):
|
|
if macro.get("type") != "sequence":
|
|
self.macros[macro_id] = self._migrate_macro(macro)
|
|
migrated = True
|
|
|
|
if migrated:
|
|
self.save_macros()
|
|
print("Migrated macros to new command sequence format")
|
|
|
|
except Exception as e:
|
|
print(f"Error loading macros: {e}")
|
|
self.macros = {}
|
|
|
|
def _migrate_macro(self, old_macro: dict) -> dict:
|
|
"""Convert old macro format to new command sequence format."""
|
|
if old_macro.get("type") == "sequence":
|
|
return old_macro
|
|
|
|
commands = []
|
|
modifiers = old_macro.get("modifiers", {})
|
|
|
|
if old_macro.get("type") == "text":
|
|
# Build held keys list
|
|
held_keys = []
|
|
for mod in ["ctrl", "alt", "shift"]:
|
|
if modifiers.get(mod):
|
|
held_keys.append(mod)
|
|
|
|
if held_keys:
|
|
# Use hotkey for modified text
|
|
commands.append({
|
|
"type": "hotkey",
|
|
"keys": held_keys + [old_macro.get("command", "")]
|
|
})
|
|
else:
|
|
# Plain text
|
|
commands.append({
|
|
"type": "text",
|
|
"value": old_macro.get("command", "")
|
|
})
|
|
|
|
# Add enter if requested
|
|
if modifiers.get("enter"):
|
|
commands.append({"type": "key", "value": "enter"})
|
|
|
|
elif old_macro.get("type") == "app":
|
|
commands.append({
|
|
"type": "app",
|
|
"command": old_macro.get("command", "")
|
|
})
|
|
|
|
return {
|
|
"name": old_macro.get("name", "Unnamed"),
|
|
"type": "sequence",
|
|
"commands": commands,
|
|
"category": old_macro.get("category", ""),
|
|
"image_path": old_macro.get("image_path", ""),
|
|
"last_used": old_macro.get("last_used", 0)
|
|
}
|
|
|
|
def save_macros(self):
|
|
"""Save macros to JSON file."""
|
|
try:
|
|
with open(self.data_file, "w") as file:
|
|
json.dump(self.macros, file, indent=4)
|
|
except Exception as e:
|
|
print(f"Error saving macros: {e}")
|
|
|
|
def get_sorted_macros(self, sort_by: str = "name"):
|
|
"""Get macros sorted by specified criteria."""
|
|
macro_list = list(self.macros.items())
|
|
|
|
if sort_by == "name":
|
|
macro_list.sort(key=lambda x: x[1]["name"].lower())
|
|
elif sort_by == "type":
|
|
# Sort by first command type in sequence
|
|
def get_first_type(macro):
|
|
cmds = macro.get("commands", [])
|
|
return cmds[0].get("type", "") if cmds else ""
|
|
macro_list.sort(key=lambda x: (get_first_type(x[1]), x[1]["name"].lower()))
|
|
elif sort_by == "recent":
|
|
macro_list.sort(
|
|
key=lambda x: (x[1].get("last_used", 0), x[1]["name"].lower()),
|
|
reverse=True
|
|
)
|
|
|
|
return macro_list
|
|
|
|
def filter_macros_by_tab(self, macro_list: list, tab_name: str):
|
|
"""Filter macros based on tab/category name."""
|
|
if tab_name == "All":
|
|
return macro_list
|
|
|
|
filtered = []
|
|
for macro_id, macro in macro_list:
|
|
# Check category match
|
|
if macro.get("category") == tab_name:
|
|
filtered.append((macro_id, macro))
|
|
# Check first command type match
|
|
elif macro.get("commands"):
|
|
first_type = macro["commands"][0].get("type", "").title()
|
|
if first_type == tab_name:
|
|
filtered.append((macro_id, macro))
|
|
|
|
return filtered
|
|
|
|
def get_unique_tabs(self) -> list:
|
|
"""Get list of unique tabs based on categories."""
|
|
tabs = ["All"]
|
|
categories = set()
|
|
|
|
for macro in self.macros.values():
|
|
if macro.get("category"):
|
|
categories.add(macro["category"])
|
|
|
|
for category in sorted(categories):
|
|
if category and category not in tabs:
|
|
tabs.append(category)
|
|
|
|
return tabs
|
|
|
|
def add_macro(
|
|
self,
|
|
name: str,
|
|
commands: list,
|
|
category: str = "",
|
|
image_path: str = ""
|
|
) -> str:
|
|
"""Add a new macro with command sequence."""
|
|
macro_id = str(uuid.uuid4())
|
|
|
|
# Process image if provided
|
|
image_path_reference = self._process_image(image_path) if image_path else ""
|
|
|
|
# Create macro data (deep copy commands to avoid reference issues)
|
|
macro_data = {
|
|
"name": name,
|
|
"type": "sequence",
|
|
"commands": copy.deepcopy(commands),
|
|
"category": category,
|
|
"image_path": image_path_reference,
|
|
"last_used": 0
|
|
}
|
|
|
|
self.macros[macro_id] = macro_data
|
|
self.save_macros()
|
|
return macro_id
|
|
|
|
def update_macro(
|
|
self,
|
|
macro_id: str,
|
|
name: str,
|
|
commands: list,
|
|
category: str = "",
|
|
image_path: Optional[str] = None
|
|
) -> bool:
|
|
"""Update an existing macro."""
|
|
if macro_id not in self.macros:
|
|
return False
|
|
|
|
macro = self.macros[macro_id]
|
|
|
|
# Keep old image or update with new one
|
|
if image_path is not None:
|
|
image_path_reference = self._process_image(image_path) if image_path else ""
|
|
else:
|
|
image_path_reference = macro.get("image_path", "")
|
|
|
|
# Update macro data (deep copy commands to avoid reference issues)
|
|
self.macros[macro_id] = {
|
|
"name": name,
|
|
"type": "sequence",
|
|
"commands": copy.deepcopy(commands),
|
|
"category": category,
|
|
"image_path": image_path_reference,
|
|
"last_used": macro.get("last_used", 0)
|
|
}
|
|
|
|
self.save_macros()
|
|
return True
|
|
|
|
def _process_image(self, image_path: str) -> str:
|
|
"""Process and store an image for a macro."""
|
|
if not image_path:
|
|
return ""
|
|
|
|
try:
|
|
file_ext = os.path.splitext(image_path)[1].lower()
|
|
unique_filename = f"{uuid.uuid4().hex}{file_ext}"
|
|
dest_path = os.path.join(self.images_dir, unique_filename)
|
|
|
|
# Resize image to max 256x256
|
|
with Image.open(image_path) as img:
|
|
img.thumbnail((256, 256))
|
|
img.save(dest_path)
|
|
|
|
return os.path.join("macro_images", unique_filename)
|
|
except Exception as e:
|
|
print(f"Error processing image: {e}")
|
|
return ""
|
|
|
|
def delete_macro(self, macro_id: str) -> bool:
|
|
"""Delete a macro."""
|
|
if macro_id not in self.macros:
|
|
return False
|
|
|
|
macro = self.macros[macro_id]
|
|
|
|
# Delete associated image file
|
|
if macro.get("image_path"):
|
|
try:
|
|
img_path = os.path.join(self.app_dir, macro["image_path"])
|
|
if os.path.exists(img_path):
|
|
os.remove(img_path)
|
|
except Exception as e:
|
|
print(f"Error removing image file: {e}")
|
|
|
|
del self.macros[macro_id]
|
|
self.save_macros()
|
|
return True
|
|
|
|
def execute_macro(self, macro_id: str) -> bool:
|
|
"""Execute a macro by ID."""
|
|
if macro_id not in self.macros:
|
|
return False
|
|
|
|
macro = self.macros[macro_id]
|
|
|
|
# Update last_used timestamp
|
|
self.macros[macro_id]["last_used"] = time.time()
|
|
self.save_macros()
|
|
|
|
try:
|
|
commands = macro.get("commands", [])
|
|
for cmd in commands:
|
|
self._execute_command(cmd)
|
|
return True
|
|
except Exception as e:
|
|
print(f"Error executing macro: {e}")
|
|
return False
|
|
|
|
def _execute_command(self, cmd: dict):
|
|
"""Execute a single command from a sequence."""
|
|
cmd_type = cmd.get("type", "")
|
|
|
|
if cmd_type == "text":
|
|
# Type text string using clipboard for Unicode support
|
|
value = cmd.get("value", "")
|
|
if value:
|
|
try:
|
|
import pyperclip
|
|
# Save current clipboard, paste text, restore (optional)
|
|
pyperclip.copy(value)
|
|
pyautogui.hotkey("ctrl", "v")
|
|
time.sleep(0.05) # Small delay after paste
|
|
except ImportError:
|
|
# Fallback to typewrite for ASCII-only
|
|
if len(value) == 1:
|
|
pyautogui.press(value)
|
|
else:
|
|
pyautogui.typewrite(value, interval=0.02)
|
|
|
|
elif cmd_type == "key":
|
|
# Press a single key
|
|
key = cmd.get("value", "")
|
|
if key:
|
|
pyautogui.press(key)
|
|
|
|
elif cmd_type == "hotkey":
|
|
# Press key combination
|
|
keys = cmd.get("keys", [])
|
|
if keys:
|
|
# Ensure keys is a list, not a string
|
|
if isinstance(keys, str):
|
|
keys = [k.strip().lower() for k in keys.split(",")]
|
|
# Small delay before hotkey for reliability on Windows
|
|
time.sleep(0.05)
|
|
pyautogui.hotkey(*keys, interval=0.05)
|
|
|
|
elif cmd_type == "wait":
|
|
# Delay in milliseconds
|
|
ms = cmd.get("ms", 0)
|
|
if ms > 0:
|
|
time.sleep(ms / 1000.0)
|
|
|
|
elif cmd_type == "app":
|
|
# Launch application.
|
|
# SECURITY: shell=True was removed to kill shell-metacharacter
|
|
# injection (no ; && | $() chaining). We tokenize the command
|
|
# and exec the program directly without a shell.
|
|
command = cmd.get("command", "")
|
|
if command:
|
|
try:
|
|
args = shlex.split(command, posix=(os.name != "nt"))
|
|
if args:
|
|
subprocess.Popen(args)
|
|
except Exception as e:
|
|
print(f"Error launching app command: {e}")
|
|
|
|
# Legacy API compatibility methods
|
|
def add_macro_legacy(
|
|
self,
|
|
name: str,
|
|
macro_type: str,
|
|
command: str,
|
|
category: str = "",
|
|
modifiers: Optional[dict] = None,
|
|
image_path: str = ""
|
|
) -> str:
|
|
"""Add macro using legacy format (auto-converts to sequence)."""
|
|
if modifiers is None:
|
|
modifiers = {"ctrl": False, "alt": False, "shift": False, "enter": False}
|
|
|
|
# Build command sequence
|
|
commands = []
|
|
held_keys = []
|
|
for mod in ["ctrl", "alt", "shift"]:
|
|
if modifiers.get(mod):
|
|
held_keys.append(mod)
|
|
|
|
if macro_type == "text":
|
|
if held_keys:
|
|
commands.append({"type": "hotkey", "keys": held_keys + [command]})
|
|
else:
|
|
commands.append({"type": "text", "value": command})
|
|
|
|
if modifiers.get("enter"):
|
|
commands.append({"type": "key", "value": "enter"})
|
|
elif macro_type == "app":
|
|
commands.append({"type": "app", "command": command})
|
|
|
|
return self.add_macro(name, commands, category, image_path)
|
|
|
|
def get_macro(self, macro_id: str) -> Optional[dict]:
|
|
"""Get a macro by ID."""
|
|
return self.macros.get(macro_id)
|
|
|
|
def get_all_macros(self) -> dict:
|
|
"""Get all macros."""
|
|
return self.macros.copy()
|