From 621e137d9ba6c9f4148aef7a869d7b38b6528e76 Mon Sep 17 00:00:00 2001 From: malmert Date: Wed, 28 Jan 2026 19:53:25 +0100 Subject: [PATCH] fix status 3 --- backend/src/api/packet_scripting_api.py | 51 ++++++++++++++++--------- 1 file changed, 32 insertions(+), 19 deletions(-) diff --git a/backend/src/api/packet_scripting_api.py b/backend/src/api/packet_scripting_api.py index 578a589..c7af38f 100644 --- a/backend/src/api/packet_scripting_api.py +++ b/backend/src/api/packet_scripting_api.py @@ -123,14 +123,20 @@ def get_unit_fragment_path(service_name: str) -> Optional[str]: def parse_unit_execstart(service_name: str) -> Optional[Dict]: """ - Query systemd for ExecStart (most reliable) and parse the script name and qnum. - Handles both plain ExecStart values and systemd's structured values that include argv[] entries. + Query systemd for ExecStart and parse the script name and qnum. + Handles: + - plain ExecStart strings + - systemd structured blobs like: + { path=... ; argv[]=/usr/bin/python /srv/fw-scripts/foo.py 5 ; ... } + - argv[] values containing spaces/newlines and quoted entries Returns dict with keys: service, exec_start, name, script_path, qnum, extra """ unit = _systemctl_unit_name(service_name) try: - p = subprocess.run(["systemctl", "show", "-p", "ExecStart", "--value", unit], - capture_output=True, text=True, check=False, timeout=2) + p = subprocess.run( + ["systemctl", "show", "-p", "ExecStart", "--value", unit], + capture_output=True, text=True, check=False, timeout=2 + ) exec_start_raw = (p.stdout or "").strip() except Exception: logger.exception("systemctl show ExecStart failed for %s", service_name) @@ -139,34 +145,40 @@ def parse_unit_execstart(service_name: str) -> Optional[Dict]: if not exec_start_raw: return {"service": service_name, "exec_start": None, "name": None, "script_path": None, "qnum": None, "extra": None} - # systemd may return a structured blob like: - # { path=/usr/bin/python3 ; argv[]=/usr/bin/python3 /srv/fw-scripts/icmp_drop.py 5 ; ... } - # or a plain commandline. We try to extract argv[] entries if present. exec_start = exec_start_raw - # detect structured form starting with '{' and containing 'argv[]=' - if exec_start_raw.lstrip().startswith("{") and "argv[]=" in exec_start_raw: - # extract all argv[] values (non-greedy until ';' or end) - argv_matches = re.findall(r'argv\[\]=(.*?)(?:\s*;|\s*$)', exec_start_raw) - # strip and join; entries may already contain spaces (they are quoted or raw), keep as-is - argv_parts = [m.strip() for m in argv_matches if m and m.strip()] + + # If structured form with argv[] entries, extract and reconstruct the argv commandline. + if "argv[]=" in exec_start_raw: + # Capture argv[]=... entries up to the next semicolon (allow newlines). + # Accept quoted arguments too: "..." or '...' + argv_matches = re.findall(r'argv\[\]=(("(?:\\.|[^"])*"|\'(?:\\.|[^\'])*\'|[^;]*?))(?:\s*;|\s*$)', exec_start_raw, flags=re.DOTALL) + argv_parts = [] + for m in argv_matches: + val = m[0] + if not val: + continue + v = val.strip() + # strip surrounding quotes if present + if (v.startswith('"') and v.endswith('"')) or (v.startswith("'") and v.endswith("'")): + v = v[1:-1] + argv_parts.append(v) if argv_parts: - # join with space to reconstruct commandline + # join argv entries with space to form a single commandline exec_start = " ".join(argv_parts) else: - # fallback to raw + # fallback: remove braces and keep raw exec_start = exec_start_raw - # Now exec_start is either a reconstructed argv[]-based commandline or the raw value. - # Find the script path under /srv/fw-scripts + # Now try to find the script path under /srv/fw-scripts m = re.search(r'(/[^ \t\n\r"]*/srv/fw-scripts/(?P[A-Za-z0-9_.-]+)\.py)\b', exec_start) if not m: - # Return raw ExecStart but no parsed metadata + # return raw exec_start but no parsed metadata return {"service": service_name, "exec_start": exec_start, "name": None, "script_path": None, "qnum": None, "extra": None} script_path = m.group(1) name = m.group("name") - # Find integer queue number after the script path (first integer token after the script) + # Find integer qnum after the script path (first integer token) after = exec_start[m.end():].strip() qnum = None extra = None @@ -184,6 +196,7 @@ def parse_unit_execstart(service_name: str) -> Optional[Dict]: return {"service": service_name, "exec_start": exec_start, "name": name, "script_path": script_path, "qnum": qnum, "extra": extra} + def is_unit_active(service_name: str) -> bool: unit = _systemctl_unit_name(service_name) p = subprocess.run(["systemctl", "is-active", "--quiet", unit])