serversdown/series3-watcher
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 4 Wiki Activity

Feat: Update settings tab implemented.

Browse Source 
Auto-updates now configurable (URL, source (gitea or private server), log activity for auto updates.
fix: Update now hardened to prevent installation of corrupt or incorrect .exe files. (security to be hardened in the future)
This commit is contained in:
serversdwn
2026-03-17 21:08:37 -04:00
parent d2a8c2d928
commit 439feb9942
7 changed files with 259 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files

156
series3_tray.py
View File

@@ -1,5 +1,5 @@
"""
Series 3 Watcher — System Tray Launcher v1.4.2
Series 3 Watcher — System Tray Launcher v1.4.3
Requires: pystray, Pillow, tkinter (stdlib)
Run with: pythonw series3_tray.py (no console window)
@@ -16,6 +16,7 @@ import sys
import subprocess
import tempfile
import threading
import configparser
import urllib.request
import urllib.error
from datetime import datetime
@@ -52,11 +53,24 @@ def _version_tuple(v):
return tuple(parts)
def check_for_update():
"""
Query Gitea for the latest release.
Returns (tag, download_url) if an update is available, else (None, None).
"""
def _update_log(msg):
"""Append a timestamped line to the watcher log for update events."""
try:
log_path = os.path.join(
os.environ.get("LOCALAPPDATA") or os.environ.get("APPDATA") or "",
"Series3Watcher", "agent_logs", "series3_watcher.log"
)
os.makedirs(os.path.dirname(log_path), exist_ok=True)
with open(log_path, "a") as f:
f.write("[{}] [updater] {}\n".format(
datetime.now().strftime("%Y-%m-%d %H:%M:%S"), msg
))
except Exception:
pass
def _check_for_update_gitea():
"""Query Gitea API for latest release. Returns (tag, download_url) or (None, None)."""
import json as _json
try:
req = urllib.request.Request(
@@ -71,21 +85,78 @@ def check_for_update():
tag = latest.get("tag_name", "")
if _version_tuple(tag) <= _version_tuple(_CURRENT_VERSION):
return None, None
# Find the .exe asset
assets = latest.get("assets", [])
for asset in assets:
name = asset.get("name", "")
if name.lower().endswith(".exe"):
name = asset.get("name", "").lower()
if name.endswith(".exe") and "setup" not in name:
return tag, asset.get("browser_download_url")
_update_log("Newer release {} found but no valid .exe asset".format(tag))
return tag, None
except Exception:
except Exception as e:
_update_log("check_for_update (gitea) failed: {}".format(e))
return None, None
def _check_for_update_url(base_url):
"""Query a custom URL server for latest version. Returns (tag, download_url) or (None, None)."""
if not base_url:
_update_log("UPDATE_SOURCE=url but UPDATE_URL is empty — skipping")
return None, None
try:
ver_url = base_url.rstrip("/") + "/api/updates/series3-watcher/version.txt"
req = urllib.request.Request(
ver_url,
headers={"User-Agent": "series3-watcher/{}".format(_CURRENT_VERSION)},
)
with urllib.request.urlopen(req, timeout=8) as resp:
tag = resp.read().decode("utf-8").strip()
if not tag:
return None, None
if _version_tuple(tag) <= _version_tuple(_CURRENT_VERSION):
return None, None
exe_url = base_url.rstrip("/") + "/api/updates/series3-watcher/series3-watcher.exe"
return tag, exe_url
except Exception as e:
_update_log("check_for_update (url mode) failed: {}".format(e))
return None, None
def check_for_update():
"""
Check for an update using the configured source (gitea, url, or disabled).
Reads UPDATE_SOURCE and UPDATE_URL from config.ini at check time.
Returns (tag, download_url) if an update is available, else (None, None).
Returns (None, None) immediately if UPDATE_SOURCE = disabled.
"""
try:
cp = configparser.ConfigParser(inline_comment_prefixes=(";", "#"))
cp.optionxform = str
cp.read(CONFIG_PATH, encoding="utf-8")
section = cp["agent"] if cp.has_section("agent") else {}
update_source = section.get("UPDATE_SOURCE", "gitea").strip().lower()
update_url = section.get("UPDATE_URL", "").strip()
except Exception:
update_source = "gitea"
update_url = ""
if update_source == "disabled":
return None, None
_update_log("Checking for update (source={}, version={})".format(
update_source, _CURRENT_VERSION
))
if update_source == "url":
return _check_for_update_url(update_url)
else:
return _check_for_update_gitea()
def apply_update(download_url):
"""
Download new .exe to a temp file, write a swap .bat, launch it, exit.
The bat waits for us to exit, then swaps the files and relaunches.
Download new .exe to a temp file, validate it, write a swap .bat, launch it, exit.
The bat backs up the old exe, retries the copy up to 5 times if locked, then relaunches.
The .exe.old backup is left in place as a rollback copy.
"""
exe_path = os.path.abspath(sys.executable if getattr(sys, "frozen", False) else sys.argv[0])
@@ -93,6 +164,8 @@ def apply_update(download_url):
tmp_fd, tmp_path = tempfile.mkstemp(suffix=".exe", prefix="s3w_update_")
os.close(tmp_fd)
_update_log("Downloading update from: {}".format(download_url))
req = urllib.request.Request(
download_url,
headers={"User-Agent": "series3-watcher/{}".format(_CURRENT_VERSION)},
@@ -101,26 +174,79 @@ def apply_update(download_url):
with open(tmp_path, "wb") as f:
f.write(resp.read())
# Three-layer validation before touching the live exe
try:
dl_size = os.path.getsize(tmp_path)
current_size = os.path.getsize(exe_path)
_update_log("Download complete ({} bytes), validating...".format(dl_size))
if dl_size < 100 * 1024:
_update_log("Validation failed: too small ({} bytes) — aborting".format(dl_size))
os.remove(tmp_path)
return False
if current_size > 0 and dl_size < current_size * 0.5:
_update_log("Validation failed: suspiciously small ({} bytes vs current {} bytes) — aborting".format(
dl_size, current_size
))
os.remove(tmp_path)
return False
with open(tmp_path, "rb") as _f:
magic = _f.read(2)
if magic != b"MZ":
_update_log("Validation failed: not a valid Windows exe (bad magic bytes) — aborting")
os.remove(tmp_path)
return False
_update_log("Validation passed ({} bytes, MZ ok)".format(dl_size))
except Exception as e:
_update_log("Validation error: {} — aborting".format(e))
try:
os.remove(tmp_path)
except Exception:
pass
return False
bat_fd, bat_path = tempfile.mkstemp(suffix=".bat", prefix="s3w_swap_")
os.close(bat_fd)
bat_content = (
"@echo off\r\n"
"ping 127.0.0.1 -n 4 > nul\r\n"
"copy /Y \"{exe}\" \"{exe}.old\"\r\n"
"set RETRIES=0\r\n"
":retry\r\n"
"copy /Y \"{new}\" \"{exe}\"\r\n"
"if errorlevel 1 (\r\n"
" set /a RETRIES+=1\r\n"
" if %RETRIES% GEQ 5 goto fail\r\n"
" ping 127.0.0.1 -n 3 > nul\r\n"
" goto retry\r\n"
")\r\n"
"start \"\" \"{exe}\"\r\n"
"del \"{new}\"\r\n"
"del \"%~f0\"\r\n"
"exit /b 0\r\n"
":fail\r\n"
"del \"{new}\"\r\n"
"del \"%~f0\"\r\n"
"exit /b 1\r\n"
).format(new=tmp_path, exe=exe_path)
with open(bat_path, "w") as f:
f.write(bat_content)
_update_log("Launching swap bat — exiting for update")
subprocess.Popen(
["cmd", "/C", bat_path],
creationflags=subprocess.CREATE_NO_WINDOW if hasattr(subprocess, "CREATE_NO_WINDOW") else 0,
)
return True
except Exception:
except Exception as e:
_update_log("apply_update failed: {}".format(e))
return False
@@ -352,7 +478,7 @@ class WatcherTray:
self._do_update()
return # exit loop; swap bat will relaunch
# Periodic Gitea update check (every ~5 min)
# Periodic update check (every ~5 min)
update_check_counter += 1
if update_check_counter >= 30:
update_check_counter = 0
@@ -379,7 +505,7 @@ class WatcherTray:
self.stop_event.set()
if self._icon is not None:
self._icon.stop()
# If update failed, just keep running silently
# If update failed, keep running silently — error is in the log
# --- Entry point ---