214 lines
8 KiB
Python
214 lines
8 KiB
Python
"""
|
|
mod_timers.py -- systemd timer installation and removal.
|
|
|
|
Manages the blocklist refresh timer, DDNS maintenance timer, and
|
|
generic interval timers (health check, etc.).
|
|
"""
|
|
|
|
import subprocess
|
|
|
|
import mod_shared as shared
|
|
|
|
BLIST_TIMER_NAME = f"{shared.PRODUCT_NAME}-dns-blocklist-update"
|
|
BLIST_TIMER_FILE = shared.SYSTEMD_DIR / f"{BLIST_TIMER_NAME}.timer"
|
|
BLIST_TIMER_SVC_FILE = shared.SYSTEMD_DIR / f"{BLIST_TIMER_NAME}.service"
|
|
|
|
MAINT_TIMER_NAME = f"{shared.PRODUCT_NAME}-maintenance"
|
|
MAINT_TIMER_FILE = shared.SYSTEMD_DIR / f"{MAINT_TIMER_NAME}.timer"
|
|
MAINT_TIMER_SVC_FILE = shared.SYSTEMD_DIR / f"{MAINT_TIMER_NAME}.service"
|
|
|
|
HEALTH_TIMER_NAME = f"{shared.PRODUCT_NAME}-health-check"
|
|
HEALTH_TIMER_FILE = shared.SYSTEMD_DIR / f"{HEALTH_TIMER_NAME}.timer"
|
|
HEALTH_TIMER_SVC_FILE = shared.SYSTEMD_DIR / f"{HEALTH_TIMER_NAME}.service"
|
|
HEALTH_TIMER_INTERVAL_SEC = 300
|
|
|
|
|
|
# ===================================================================
|
|
# Blocklist timer
|
|
# ===================================================================
|
|
|
|
def _parse_time_to_calendar(time_str):
|
|
"""Convert HH:MM string to a systemd OnCalendar value. Raises ValueError on bad format."""
|
|
parts = time_str.strip().split(":")
|
|
if len(parts) != 2:
|
|
raise ValueError(f"Invalid daily_execute_time_24hr_local: '{time_str}'. Expected HH:MM.")
|
|
hh, mm = parts
|
|
return f"*-*-* {hh.zfill(2)}:{mm.zfill(2)}:00"
|
|
|
|
def install_timer(data):
|
|
"""Install the daily blocklist refresh timer. Returns error string on failure, None on success."""
|
|
general = data.get("dns_blocking", {}).get("general", {})
|
|
execute_time = general.get("daily_execute_time_24hr_local", "02:30")
|
|
try:
|
|
on_calendar = _parse_time_to_calendar(execute_time)
|
|
except ValueError as e:
|
|
return str(e)
|
|
|
|
timer_content = "\n".join([
|
|
"# Generated by core.py -- do not edit manually.",
|
|
"",
|
|
"[Unit]",
|
|
"Description=Daily blocklist refresh",
|
|
"",
|
|
"[Timer]",
|
|
f"OnCalendar={on_calendar}",
|
|
"Persistent=true",
|
|
"",
|
|
"[Install]",
|
|
"WantedBy=timers.target",
|
|
"",
|
|
])
|
|
|
|
blocklist_script = shared.SCRIPT_DIR / "dns-blocklists.py"
|
|
service_content = "\n".join([
|
|
"# Generated by core.py -- do not edit manually.",
|
|
"",
|
|
"[Unit]",
|
|
"Description=Daily blocklist refresh",
|
|
"After=network-online.target",
|
|
"Wants=network-online.target",
|
|
"",
|
|
"[Service]",
|
|
"Type=oneshot",
|
|
f"ExecStart=/usr/bin/python3 {blocklist_script}",
|
|
"",
|
|
])
|
|
|
|
for path, content in ((BLIST_TIMER_FILE, timer_content), (BLIST_TIMER_SVC_FILE, service_content)):
|
|
if not path.exists() or path.read_text() != content:
|
|
path.write_text(content)
|
|
print(f"Written: {path}")
|
|
|
|
subprocess.run(["systemctl", "daemon-reload"], capture_output=True, text=True)
|
|
subprocess.run(["systemctl", "enable", "--now", f"{BLIST_TIMER_NAME}.timer"],
|
|
capture_output=True, text=True)
|
|
print(f"Timer {BLIST_TIMER_NAME}.timer enabled (runs daily at {execute_time}).")
|
|
|
|
|
|
# ===================================================================
|
|
# Interval timers
|
|
# ===================================================================
|
|
|
|
def install_interval_timers(names, timer_files, svc_files, descriptions, exec_starts, interval_secs):
|
|
for name, timer_file, svc_file, description, exec_start, interval_sec in zip(
|
|
names, timer_files, svc_files, descriptions, exec_starts, interval_secs):
|
|
timer_content = "\n".join([
|
|
"# Generated by core.py -- do not edit manually.",
|
|
"",
|
|
"[Unit]",
|
|
f"Description={description}",
|
|
"",
|
|
"[Timer]",
|
|
f"OnActiveSec={interval_sec}s",
|
|
f"OnUnitActiveSec={interval_sec}s",
|
|
"AccuracySec=10s",
|
|
"",
|
|
"[Install]",
|
|
"WantedBy=timers.target",
|
|
"",
|
|
])
|
|
service_content = "\n".join([
|
|
"# Generated by core.py -- do not edit manually.",
|
|
"",
|
|
"[Unit]",
|
|
f"Description={description}",
|
|
"",
|
|
"[Service]",
|
|
"Type=oneshot",
|
|
f"ExecStart={exec_start}",
|
|
"",
|
|
])
|
|
for path, content in ((timer_file, timer_content), (svc_file, service_content)):
|
|
if not path.exists() or path.read_text() != content:
|
|
path.write_text(content)
|
|
print(f"Written: {path}")
|
|
|
|
subprocess.run(["systemctl", "daemon-reload"], capture_output=True, text=True)
|
|
for name, interval_sec in zip(names, interval_secs):
|
|
subprocess.run(["systemctl", "enable", f"{name}.timer"], capture_output=True, text=True)
|
|
active = subprocess.run(
|
|
["systemctl", "is-active", f"{name}.timer"],
|
|
capture_output=True, text=True
|
|
).stdout.strip() == "active"
|
|
verb = "restart" if active else "start"
|
|
subprocess.run(["systemctl", verb, f"{name}.timer"], capture_output=True, text=True)
|
|
print(f"Timer {name}.timer enabled (runs every {interval_sec}s).")
|
|
|
|
def remove_timers(names, timer_files, svc_files, daemon_reload=False):
|
|
for name, timer_file, svc_file in zip(names, timer_files, svc_files):
|
|
subprocess.run(["systemctl", "disable", "--now", f"{name}.timer"],
|
|
capture_output=True, text=True)
|
|
for f in (timer_file, svc_file):
|
|
if f.exists():
|
|
f.unlink()
|
|
print(f"Removed: {f}")
|
|
else:
|
|
print(f"Not found, skipping: {f}")
|
|
if daemon_reload:
|
|
subprocess.run(["systemctl", "daemon-reload"], capture_output=True, text=True)
|
|
|
|
|
|
# ===================================================================
|
|
# DDNS maintenance timer
|
|
# ===================================================================
|
|
|
|
def _parse_ddns_interval(interval_str):
|
|
"""Convert interval string (e.g. 5m, 2h, 1d) to systemd OnUnitActiveSec value."""
|
|
s = interval_str.strip()
|
|
if s.endswith("m"): return f"{s[:-1]}min"
|
|
if s.endswith("h"): return f"{s[:-1]}h"
|
|
if s.endswith("d"): return f"{s[:-1]}day"
|
|
raise ValueError(f"Invalid timer_interval format: '{s}'. Use e.g. 5m, 2h, 1d.")
|
|
|
|
def install_maint_timer(data):
|
|
ddns = data.get("ddns", {})
|
|
interval = ddns.get("general", {}).get("timer_interval", "10m")
|
|
script_path = shared.SCRIPT_DIR / "maintenance.py"
|
|
try:
|
|
systemd_unit = _parse_ddns_interval(interval)
|
|
except ValueError as e:
|
|
print(f"DDNS timer: {e}")
|
|
return
|
|
|
|
service_content = "\n".join([
|
|
"# Generated by core.py -- do not edit manually.",
|
|
"",
|
|
"[Unit]",
|
|
"Description=DDNS IP update",
|
|
"After=network-online.target",
|
|
"Wants=network-online.target",
|
|
"",
|
|
"[Service]",
|
|
"Type=oneshot",
|
|
f"ExecStart=/usr/bin/python3 {script_path} --update",
|
|
"",
|
|
])
|
|
timer_content = "\n".join([
|
|
"# Generated by core.py -- do not edit manually.",
|
|
"",
|
|
"[Unit]",
|
|
"Description=DDNS IP update timer",
|
|
"",
|
|
"[Timer]",
|
|
f"OnActiveSec={systemd_unit}",
|
|
f"OnUnitActiveSec={systemd_unit}",
|
|
"OnBootSec=1min",
|
|
"AccuracySec=10s",
|
|
"",
|
|
"[Install]",
|
|
"WantedBy=timers.target",
|
|
"",
|
|
])
|
|
for path, content in ((MAINT_TIMER_SVC_FILE, service_content), (MAINT_TIMER_FILE, timer_content)):
|
|
if not path.exists() or path.read_text() != content:
|
|
path.write_text(content)
|
|
print(f"Written: {path}")
|
|
subprocess.run(["systemctl", "daemon-reload"], capture_output=True, text=True)
|
|
active = subprocess.run(
|
|
["systemctl", "is-active", f"{MAINT_TIMER_NAME}.timer"],
|
|
capture_output=True, text=True
|
|
).stdout.strip() == "active"
|
|
verb = "restart" if active else "enable --now"
|
|
subprocess.run(["systemctl"] + verb.split() + [f"{MAINT_TIMER_NAME}.timer"],
|
|
capture_output=True, text=True)
|
|
print(f"Timer {MAINT_TIMER_NAME}.timer enabled (runs every {interval}).")
|