feat: improve web configuration controls
This commit is contained in:
@@ -12,10 +12,13 @@ TOKEN=Path("/etc/pigway-pi-control-api.token")
|
||||
WEB=Path("/usr/local/share/pigway-pi-control/web/index.html")
|
||||
ALLOWED_SECTIONS={"services","processes","alerts","fan","led","timing","dark_mode","display","api"}
|
||||
NAME_RE=re.compile(r"^[A-Za-z0-9_.-]{1,64}$")
|
||||
UNIT_RE=re.compile(r"^[A-Za-z0-9_.@:-]{1,128}\.service$")
|
||||
RGB_KEY_RE=re.compile(r"^(cpu|power|memory|storage|network|service|normal)_[rgb]$")
|
||||
FLASH_KEY_RE=re.compile(r"^(warning|critical|emergency|normal)_flash_(on|off)_ms$")
|
||||
NORMAL_MODES={"off","solid","flash","flow","breathe","marquee","rainbow","colorful"}
|
||||
EFFECT_COLORS={"red","green","blue","yellow","purple","cyan","white"}
|
||||
PROTECTED_UNITS={"ssh.service","sshd.service","runawesun.service","tailscaled.service",
|
||||
"pigway-pi-control.service","pigway-pi-control-api.service"}
|
||||
|
||||
def load_cfg():
|
||||
cfg=configparser.ConfigParser(); cfg.read(CFG)
|
||||
@@ -102,6 +105,76 @@ def journal(limit):
|
||||
except json.JSONDecodeError: pass
|
||||
return rows
|
||||
|
||||
def service_inventory():
|
||||
cfg=load_cfg()
|
||||
monitored={target.lower():(name,target) for name,target in cfg.items("services")} if cfg.has_section("services") else {}
|
||||
notes=dict(cfg.items("service_notes")) if cfg.has_section("service_notes") else {}
|
||||
files=subprocess.run(["systemctl","list-unit-files","--type=service","--no-legend","--no-pager"],
|
||||
capture_output=True,text=True,timeout=10,check=True)
|
||||
units={}
|
||||
for line in files.stdout.splitlines():
|
||||
parts=line.split()
|
||||
if len(parts)>=2 and UNIT_RE.fullmatch(parts[0]):
|
||||
units[parts[0]]={"unit":parts[0],"enabled":parts[1],"active":"inactive","description":""}
|
||||
states=subprocess.run(["systemctl","list-units","--all","--type=service","--no-legend","--no-pager","--plain"],
|
||||
capture_output=True,text=True,timeout=10,check=True)
|
||||
for line in states.stdout.splitlines():
|
||||
parts=line.split(None,4)
|
||||
if len(parts)>=4 and UNIT_RE.fullmatch(parts[0]):
|
||||
item=units.setdefault(parts[0],{"unit":parts[0],"enabled":"unknown"})
|
||||
item.update({"active":parts[2],"description":parts[4] if len(parts)>4 else ""})
|
||||
for name,target in monitored.values():
|
||||
item=units.setdefault(target,{"unit":target,"enabled":"not-found","active":"inactive","description":""})
|
||||
item.update({"monitored":True,"monitor_name":name.upper(),"note":notes.get(name,"")})
|
||||
for item in units.values():
|
||||
item.setdefault("monitored",False); item.setdefault("monitor_name",""); item.setdefault("note","")
|
||||
item["protected"]=item["unit"] in PROTECTED_UNITS
|
||||
return sorted(units.values(),key=lambda x:(not x["monitored"],x["unit"]))
|
||||
|
||||
def edit_service_monitor(operation,name,target,note="",previous_name=""):
|
||||
if operation not in {"save","delete"}: raise ValueError("operation must be save or delete")
|
||||
if not NAME_RE.fullmatch(name): raise ValueError("invalid monitor name")
|
||||
if not UNIT_RE.fullmatch(target): raise ValueError("invalid systemd service name")
|
||||
if len(note)>120 or any(c in note for c in "\r\n\x00"): raise ValueError("invalid note")
|
||||
if previous_name and not NAME_RE.fullmatch(previous_name): raise ValueError("invalid previous monitor name")
|
||||
cfg=load_cfg(); old_target=cfg.get("services",name,fallback=None)
|
||||
text=CFG.read_text()
|
||||
def change(section,key,value):
|
||||
nonlocal text
|
||||
section_match=re.search(rf"(?mi)^\[{re.escape(section)}\]\s*$",text)
|
||||
if not section_match:
|
||||
if value is not None: text=text.rstrip()+f"\n\n[{section}]\n{key} = {value}\n"
|
||||
return
|
||||
next_section=re.search(r"(?m)^\[.+\]\s*$",text[section_match.end():])
|
||||
end=section_match.end()+(next_section.start() if next_section else len(text)-section_match.end())
|
||||
chunk=text[section_match.end():end]
|
||||
key_match=re.search(rf"(?mi)^\s*{re.escape(key)}\s*=.*(?:\n|$)",chunk)
|
||||
if key_match:
|
||||
start=section_match.end()+key_match.start(); stop=section_match.end()+key_match.end()
|
||||
text=text[:start]+((f"{key} = {value}\n") if value is not None else "")+text[stop:]
|
||||
elif value is not None:
|
||||
text=text[:end].rstrip()+f"\n{key} = {value}\n\n"+text[end:].lstrip("\n")
|
||||
if operation=="delete":
|
||||
if old_target is None: raise ValueError("monitor not found")
|
||||
change("services",name,None); change("service_notes",name,None)
|
||||
else:
|
||||
if previous_name and previous_name.lower()!=name.lower():
|
||||
change("services",previous_name,None); change("service_notes",previous_name,None)
|
||||
change("services",name,target); change("service_notes",name,note or None)
|
||||
parsed=configparser.ConfigParser(); parsed.read_string(text)
|
||||
backup=CFG.with_name(f"{CFG.name}.bak.api-{time.strftime('%Y%m%d-%H%M%S')}")
|
||||
shutil.copy2(CFG,backup)
|
||||
temporary=CFG.with_suffix(".tmp"); temporary.write_text(text); temporary.chmod(0o644); temporary.replace(CFG)
|
||||
subprocess.run(["systemctl","restart","pigway-pi-control.service"],check=True,timeout=10)
|
||||
return str(backup)
|
||||
|
||||
def control_service(unit,action):
|
||||
if not UNIT_RE.fullmatch(unit): raise ValueError("invalid systemd service name")
|
||||
if action not in {"start","stop"}: raise ValueError("action must be start or stop")
|
||||
if unit in PROTECTED_UNITS and action=="stop": raise ValueError("protected remote-management service cannot be stopped here")
|
||||
result=subprocess.run(["systemctl",action,unit],capture_output=True,text=True,timeout=20)
|
||||
if result.returncode: raise RuntimeError(result.stderr.strip() or f"systemctl {action} failed")
|
||||
|
||||
class Handler(BaseHTTPRequestHandler):
|
||||
server_version="PIGWayAPI/3.6"
|
||||
|
||||
@@ -146,14 +219,28 @@ class Handler(BaseHTTPRequestHandler):
|
||||
self.send_json(200,{"logs":journal(limit)}); return
|
||||
if parsed.path=="/api/v1/config":
|
||||
self.send_json(200,{"config":config_json(),"write_requires_token":True}); return
|
||||
if parsed.path=="/api/v1/services":
|
||||
self.send_json(200,{"services":service_inventory()}); return
|
||||
self.send_json(404,{"error":"not found"})
|
||||
except Exception as e: self.send_json(500,{"error":str(e)})
|
||||
|
||||
def do_PUT(self):
|
||||
if urlparse(self.path).path!="/api/v1/config": self.send_json(404,{"error":"not found"}); return
|
||||
path=urlparse(self.path).path
|
||||
if path not in {"/api/v1/config","/api/v1/services","/api/v1/services/control"}: self.send_json(404,{"error":"not found"}); return
|
||||
if not self.authorized(): self.send_json(401,{"error":"bearer token required"}); return
|
||||
try:
|
||||
body=self.read_json(); updates=body.get("updates")
|
||||
body=self.read_json()
|
||||
if path=="/api/v1/services/control":
|
||||
control_service(str(body.get("unit","")),str(body.get("action","")))
|
||||
print(f"SERVICE_CONTROL action={body.get('action')} unit={body.get('unit')}",flush=True)
|
||||
self.send_json(200,{"ok":True}); return
|
||||
if path=="/api/v1/services":
|
||||
backup=edit_service_monitor(str(body.get("operation","")),str(body.get("name","")),
|
||||
str(body.get("target","")),str(body.get("note","")),
|
||||
str(body.get("previous_name","")))
|
||||
print(f"SERVICE_MONITOR_UPDATED operation={body.get('operation')} target={body.get('target')} backup={backup}",flush=True)
|
||||
self.send_json(200,{"ok":True,"backup":backup}); return
|
||||
updates=body.get("updates")
|
||||
backup=update_ini(updates)
|
||||
subprocess.run(["systemctl","restart","pigway-pi-control.service"],check=True,timeout=10)
|
||||
restart_api="api" in updates
|
||||
|
||||
Reference in New Issue
Block a user