Files
pigway-device-agent/app/pigway_pi_control_api.py

399 lines
23 KiB
Python

#!/usr/bin/env python3
import sys
sys.path.append("/usr/local/share/pigway-pi-control")
from plugin_api import discover,registered,request as plugin_request,socket_path
import configparser,ipaddress,json,re,secrets,shutil,subprocess,threading,time,socket
from concurrent.futures import ThreadPoolExecutor,as_completed
from http import HTTPStatus
from http.server import BaseHTTPRequestHandler,ThreadingHTTPServer
from pathlib import Path
from urllib.parse import parse_qs,urlencode,urlparse
from urllib.request import Request,urlopen
APP_VERSION="3.7.0"
CFG=Path("/etc/pigway-pi-control.conf")
STATUS=Path("/run/pigway-pi-control/status.json")
TOKEN=Path("/etc/pigway-pi-control-api.token")
ALLOWED_SECTIONS={"device","services","processes","alerts","timing","dark_mode","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={"pigway-pi-control-api.service"}
def plugin_inventory():
cfg=load_cfg();items=discover(cfg)
for item in items:item["paused"]=cfg.getboolean("plugin_paused",item["id"],fallback=False)
return items
def find_plugin(plugin_id):
for item in plugin_inventory():
if item["id"]==plugin_id:return item
raise ValueError("unknown plugin")
def plugin_operation(body):
plugin_id=str(body.get("id",""))
if not NAME_RE.fullmatch(plugin_id):raise ValueError("invalid plugin id")
item=find_plugin(plugin_id);operation=body.get("operation")
if operation=="link":
if not item["online"]:raise ValueError("plugin is offline")
text=replace_ini_value("plugins",plugin_id,socket_path(item["socket"]))
save_ini_text(text,"plugin-link")
text=replace_ini_value("plugin_paused",plugin_id,"false")
elif operation in {"pause","resume","unlink"}:
if not item["linked"]:raise ValueError("plugin is not linked")
text=replace_ini_value("plugins",plugin_id,None) if operation=="unlink" else replace_ini_value("plugin_paused",plugin_id,"true" if operation=="pause" else "false")
else:raise ValueError("invalid plugin operation")
backup=save_ini_text(text,"plugin")
# Only reload a monitor that is already running. Never start hardware here.
subprocess.run(["systemctl","try-restart","pigway-pi-control.service"],check=True,timeout=15)
if operation in {"pause","unlink"}:
try:plugin_request(item["socket"],"DELETE","/v1/state",{})
except Exception:pass # An offline plugin expires its lease independently.
return {"ok":True,"backup":backup}
def machine_identity():
cfg=load_cfg(); identifier=cfg.get("device","identifier",fallback="auto").strip()
if not identifier or identifier.lower()=="auto": identifier=socket.gethostname()
name=cfg.get("device","name",fallback=identifier).strip() or identifier
if name.lower()=="auto": name=identifier
return {"id":identifier,"name":name,"hostname":socket.gethostname(),"local":True,"url":""}
def load_cfg():
cfg=configparser.ConfigParser(); cfg.read(CFG)
return cfg
def api_setting(key,default,cast=str):
try:return cast(load_cfg().get("api",key))
except Exception:return default
def config_json():
cfg=load_cfg()
return {section:{k:v for k,v in cfg.items(section) if not (section=="timing" and k in {"oled_refresh_interval","page_interval"})} for section in cfg.sections() if section in ALLOWED_SECTIONS}
def validate_value(section,key,value,existing):
if section=="api" and key=="enabled":
if value.lower() not in {"true","false"}:raise ValueError("enabled must be true or false")
return
if section=="timing" and key in {"oled_refresh_interval","page_interval"}:raise ValueError("display timing is managed by the hardware plugin")
if section not in {"services","processes"} and key not in existing.get(section,set()):
raise ValueError(f"unsupported option: {section}.{key}")
if section=="hardware" and key=="cooling_hat_enabled":
if value.lower() not in ("true","false"): raise ValueError("cooling_hat_enabled must be true or false")
elif section=="led" and key=="alert_mode":
if value.lower() not in ("breathe","flash"): raise ValueError("alert_mode must be breathe or flash")
elif section=="led" and key=="custom_min_hold_ms":
if not 1500<=int(value)<=60000: raise ValueError("custom_min_hold_ms must be 1500..60000")
elif RGB_KEY_RE.fullmatch(key):
number=int(value)
if not 0<=number<=255: raise ValueError(f"{key} must be 0..255")
elif FLASH_KEY_RE.fullmatch(key):
number=int(value)
if not 50<=number<=60000: raise ValueError(f"{key} must be 50..60000")
elif section=="led" and key=="normal_mode":
if value.lower() not in NORMAL_MODES: raise ValueError("normal_mode must be off, solid, flash, flow, breathe, marquee, rainbow or colorful")
elif section=="led" and key=="normal_effect_speed":
if int(value) not in (1,2,3): raise ValueError("normal_effect_speed must be 1, 2 or 3")
elif section=="led" and key=="normal_effect_color":
if value.lower() not in EFFECT_COLORS: raise ValueError("normal_effect_color must be red, green, blue, yellow, purple, cyan or white")
elif section=="api" and key=="port":
number=int(value)
if not 1<=number<=65535: raise ValueError("api.port must be 1..65535")
elif section=="api" and key=="log_limit":
number=int(value)
if not 1<=number<=1000: raise ValueError("api.log_limit must be 1..1000")
elif section=="api" and key=="bind":
ipaddress.ip_address(value)
elif section=="device" and key=="identifier" and value.lower()!="auto" and not NAME_RE.fullmatch(value):
raise ValueError("device.identifier must be auto or 1..64 letters, numbers, dot, underscore or dash")
def replace_ini_value(section,key,value):
text=CFG.read_text()
section_match=re.search(rf"(?mi)^\[{re.escape(section)}\][ \t]*$",text)
if not section_match:
if value is not None: text=text.rstrip()+f"\n\n[{section}]\n{key} = {value}\n"
return text
next_section=re.search(r"(?m)^\[[^\]\r\n]+\]",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)^[ \t]*{re.escape(key)}\s*=.*(?:\n|$)",chunk)
if key_match:
start=section_match.end()+key_match.start(); stop=section_match.end()+key_match.end()
return text[:start]+((f"{key} = {value}\n") if value is not None else "")+text[stop:]
if value is not None:return text[:end].rstrip()+f"\n{key} = {value}\n\n"+text[end:].lstrip("\n")
return text
def save_ini_text(text,prefix="api"):
parsed=configparser.ConfigParser(); parsed.read_string(text)
backup=CFG.with_name(f"{CFG.name}.bak.{prefix}-{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)
return str(backup)
def update_ini(updates):
if not isinstance(updates,dict): raise ValueError("updates must be an object")
text=CFG.read_text()
current=load_cfg(); existing={section:set(current.options(section)) for section in current.sections()}
for section,values in updates.items():
if section not in ALLOWED_SECTIONS or not isinstance(values,dict):
raise ValueError(f"unsupported section: {section}")
for key,value in values.items():
if not NAME_RE.fullmatch(str(key)): raise ValueError(f"invalid key: {key}")
if value is None:
raise ValueError(f"null is not allowed: {section}.{key}")
value=str(value).strip()
if not value or len(value)>256 or any(c in value for c in "\r\n\x00"):
raise ValueError(f"invalid value: {section}.{key}")
validate_value(section,str(key),value,existing)
section_match=re.search(rf"(?mi)^\[{re.escape(section)}\][ \t]*$",text)
if not section_match:
text=text.rstrip()+f"\n\n[{section}]\n{key} = {value}\n"
continue
next_section=re.search(r"(?m)^\[[^\]\r\n]+\]",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(str(key))}\s*=\s*).*$",chunk)
if key_match:
start=section_match.end()+key_match.start(); stop=section_match.end()+key_match.end()
replacement=key_match.group(1)+value
text=text[:start]+replacement+text[stop:]
else:
text=text[:end].rstrip()+f"\n{key} = {value}\n\n"+text[end:].lstrip("\n")
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)
return str(backup)
def journal_query(since=None,until=None,severity="",event="",search="",sort="timestamp",order="desc",limit=100,offset=0):
command=["journalctl","-u","pigway-pi-control.service","--no-pager","-o","json"]
if since is not None: command.extend(["--since",f"@{since}"])
if until is not None: command.extend(["--until",f"@{until}"])
result=subprocess.run(command,capture_output=True,text=True,timeout=15)
if result.returncode: raise RuntimeError(result.stderr.strip() or "journalctl failed")
rows=[]
for line in result.stdout.splitlines():
try:
item=json.loads(line)
message=item.get("MESSAGE","")
level_match=re.search(r"(?:^|\s)level=([A-Z]+)(?:\s|$)",message)
event_match=re.search(r"(?:^|\s)event=([^ ]+)",message)
priority=str(item.get("PRIORITY","6"))
level=(level_match.group(1) if level_match else
{"0":"EMERGENCY","1":"ALERT","2":"CRITICAL","3":"ERROR","4":"WARN","5":"NOTICE","6":"INFO","7":"DEBUG"}.get(priority,"INFO"))
rows.append({"timestamp":int(item.get("__REALTIME_TIMESTAMP",0)),"priority":priority,
"severity":level,"event":event_match.group(1) if event_match else "SYSTEM",
"message":message})
except json.JSONDecodeError: pass
severities={x.strip().upper() for x in severity.split(",") if x.strip()}
if severities: rows=[x for x in rows if x["severity"] in severities]
event_types=sorted({x["event"] for x in rows})
if event: rows=[x for x in rows if x["event"]==event]
if search:
needle=search.casefold(); rows=[x for x in rows if needle in x["message"].casefold()]
severity_rank={"EMERGENCY":0,"ALERT":1,"CRITICAL":2,"ERROR":3,"WARN":4,"NOTICE":5,"INFO":6,"DEBUG":7}
keys={"timestamp":lambda x:x["timestamp"],"severity":lambda x:severity_rank.get(x["severity"],99),
"event":lambda x:x["event"],"message":lambda x:x["message"]}
rows.sort(key=keys[sort],reverse=order=="desc")
total=len(rows)
return {"logs":rows[offset:offset+limit],"total":total,"offset":offset,"limit":limit,
"event_types":event_types,"retention":"systemd-journal"}
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":""})
paused=cfg.getboolean("service_monitor_disabled",name,fallback=False)
item.update({"monitored":True,"paused":paused,"monitor_name":name.upper(),"note":notes.get(name,"")})
for item in units.values():
item.setdefault("monitored",False); item.setdefault("paused",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","pause","resume"}: raise ValueError("invalid monitor operation")
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)
previous=previous_name or name
was_paused=cfg.getboolean("service_monitor_disabled",previous,fallback=False)
text=CFG.read_text()
def change(section,key,value):
nonlocal text
section_match=re.search(rf"(?mi)^\[{re.escape(section)}\][ \t]*$",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)^\[[^\]\r\n]+\]",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)^[ \t]*{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); change("service_monitor_disabled",name,None)
elif operation in {"pause","resume"}:
if old_target is None: raise ValueError("monitor not found")
change("service_monitor_disabled",name,"true" if operation=="pause" else None)
else:
if previous_name and previous_name.lower()!=name.lower():
change("services",previous_name,None); change("service_notes",previous_name,None)
change("service_monitor_disabled",previous_name,None)
change("services",name,target); change("service_notes",name,note or None)
if was_paused: change("service_monitor_disabled",name,"true")
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.7"
def log_message(self,fmt,*args):
return
def send_json(self,status,payload):
data=json.dumps(payload,ensure_ascii=False,separators=(",",":")).encode()
self.send_response(status); self.send_header("Content-Type","application/json; charset=utf-8")
self.send_header("Content-Length",str(len(data))); self.security_headers(); self.end_headers(); self.wfile.write(data)
def security_headers(self):
self.send_header("Cache-Control","no-store")
self.send_header("X-Content-Type-Options","nosniff")
self.send_header("X-Frame-Options","DENY")
self.send_header("Content-Security-Policy","default-src 'self'; style-src 'self' 'unsafe-inline'; script-src 'self' 'unsafe-inline'")
def authorized(self):
try: expected=TOKEN.read_text().strip()
except OSError: return False
supplied=self.headers.get("Authorization","")
return supplied.startswith("Bearer ") and secrets.compare_digest(supplied[7:],expected)
def read_json(self):
length=int(self.headers.get("Content-Length","0"))
if length<=0 or length>65536: raise ValueError("invalid request size")
return json.loads(self.rfile.read(length))
def do_GET(self):
parsed=urlparse(self.path)
try:
if not self.authorized():self.send_json(401,{"error":"bearer token required"});return
if parsed.path=="/api/v1/health":
self.send_json(200,{"ok":True,"version":APP_VERSION,"status_available":STATUS.exists()});return
if parsed.path=="/api/v1/status":
if not STATUS.exists(): self.send_json(503,{"error":"agent status unavailable"}); return
self.send_json(200,json.loads(STATUS.read_text())); return
if parsed.path=="/api/v1/logs":
query=parse_qs(parsed.query)
value=lambda key,default="":query.get(key,[default])[0]
limit=min(max(int(value("limit",api_setting("log_limit",200,int))),1),50000)
offset=max(int(value("offset",0)),0)
since=int(value("since")) if value("since") else None
until=int(value("until")) if value("until") else None
if since is not None and until is not None and since>until: raise ValueError("since must not be after until")
sort=value("sort","timestamp"); order=value("order","desc")
allowed_sort={"timestamp","severity","event","message"}
if sort not in allowed_sort: raise ValueError("invalid log sort")
if order not in {"asc","desc"}: raise ValueError("invalid log order")
severity=value("severity").upper(); event=value("event"); search=value("search")
if len(event)>80 or len(search)>120: raise ValueError("log filter is too long")
self.send_json(200,journal_query(since,until,severity,event,search,sort,order,limit,offset)); return
if parsed.path in {"/api/v1/plugins","/api/v1/plugin/config"}:
query=parse_qs(parsed.query);host_id=query.get("host",[""])[0]
plugin_id=query.get("id",[""])[0]
if parsed.path.endswith("/plugins"):
self.send_json(200,{"plugins":plugin_inventory()});return
item=find_plugin(plugin_id)
self.send_json(200,plugin_request(item["socket"],"GET","/v1/config"));return
if parsed.path=="/api/v1/config":
query=parse_qs(parsed.query); host_id=query.get("host",[""])[0]
self.send_json(200,{"config":config_json(),"write_requires_token":True,"machine":machine_identity()}); return
if parsed.path=="/api/v1/services":
query=parse_qs(parsed.query); host_id=query.get("host",[""])[0]
self.send_json(200,{"services":service_inventory(),"machine":machine_identity()}); return
self.send_json(404,{"error":"not found"})
except ValueError as e: self.send_json(400,{"error":str(e)})
except Exception as e: self.send_json(500,{"error":str(e)})
def do_PUT(self):
path=urlparse(self.path).path
if path not in {"/api/v1/plugins","/api/v1/plugin/config","/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()
if path in {"/api/v1/plugins","/api/v1/plugin/config"}:
host_id=str(body.get("host",""))
if path.endswith("/plugins"):self.send_json(200,plugin_operation(body));return
item=find_plugin(str(body.get("id","")))
self.send_json(200,plugin_request(item["socket"],"PUT","/v1/config",{"updates":body.get("updates")}));return
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
print(f"CONFIG_UPDATED sections={','.join(sorted(updates))} backup={backup}",flush=True)
self.send_json(200,{"ok":True,"backup":backup,"api_restart":restart_api})
if restart_api: threading.Timer(0.2,self.server.shutdown).start()
except (ValueError,json.JSONDecodeError) as e: self.send_json(400,{"error":str(e)})
except Exception as e: self.send_json(500,{"error":str(e)})
def main():
if not load_cfg().getboolean("api","enabled",fallback=False):
print("API_DISABLED",flush=True);sys.exit(78)
host=api_setting("bind","0.0.0.0")
port=api_setting("port",6001,int)
server=ThreadingHTTPServer((host,port),Handler)
print(f"API_START version={APP_VERSION} bind={host} port={port}",flush=True)
try: server.serve_forever()
finally: server.server_close()
if __name__=="__main__": main()