fix status 3
All checks were successful
Build and Deploy MITM Webserver / build (push) Successful in 9s
All checks were successful
Build and Deploy MITM Webserver / build (push) Successful in 9s
This commit is contained in:
@@ -123,14 +123,20 @@ def get_unit_fragment_path(service_name: str) -> Optional[str]:
|
|||||||
|
|
||||||
def parse_unit_execstart(service_name: str) -> Optional[Dict]:
|
def parse_unit_execstart(service_name: str) -> Optional[Dict]:
|
||||||
"""
|
"""
|
||||||
Query systemd for ExecStart (most reliable) and parse the script name and qnum.
|
Query systemd for ExecStart and parse the script name and qnum.
|
||||||
Handles both plain ExecStart values and systemd's structured values that include argv[] entries.
|
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
|
Returns dict with keys: service, exec_start, name, script_path, qnum, extra
|
||||||
"""
|
"""
|
||||||
unit = _systemctl_unit_name(service_name)
|
unit = _systemctl_unit_name(service_name)
|
||||||
try:
|
try:
|
||||||
p = subprocess.run(["systemctl", "show", "-p", "ExecStart", "--value", unit],
|
p = subprocess.run(
|
||||||
capture_output=True, text=True, check=False, timeout=2)
|
["systemctl", "show", "-p", "ExecStart", "--value", unit],
|
||||||
|
capture_output=True, text=True, check=False, timeout=2
|
||||||
|
)
|
||||||
exec_start_raw = (p.stdout or "").strip()
|
exec_start_raw = (p.stdout or "").strip()
|
||||||
except Exception:
|
except Exception:
|
||||||
logger.exception("systemctl show ExecStart failed for %s", service_name)
|
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:
|
if not exec_start_raw:
|
||||||
return {"service": service_name, "exec_start": None, "name": None, "script_path": None, "qnum": None, "extra": None}
|
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 argv_parts:
|
|
||||||
# join with space to reconstruct commandline
|
|
||||||
exec_start = " ".join(argv_parts)
|
|
||||||
else:
|
|
||||||
# fallback to raw
|
|
||||||
exec_start = exec_start_raw
|
exec_start = exec_start_raw
|
||||||
|
|
||||||
# Now exec_start is either a reconstructed argv[]-based commandline or the raw value.
|
# If structured form with argv[] entries, extract and reconstruct the argv commandline.
|
||||||
# Find the script path under /srv/fw-scripts
|
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 argv entries with space to form a single commandline
|
||||||
|
exec_start = " ".join(argv_parts)
|
||||||
|
else:
|
||||||
|
# fallback: remove braces and keep raw
|
||||||
|
exec_start = exec_start_raw
|
||||||
|
|
||||||
|
# Now try to find the script path under /srv/fw-scripts
|
||||||
m = re.search(r'(/[^ \t\n\r"]*/srv/fw-scripts/(?P<name>[A-Za-z0-9_.-]+)\.py)\b', exec_start)
|
m = re.search(r'(/[^ \t\n\r"]*/srv/fw-scripts/(?P<name>[A-Za-z0-9_.-]+)\.py)\b', exec_start)
|
||||||
if not m:
|
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}
|
return {"service": service_name, "exec_start": exec_start, "name": None, "script_path": None, "qnum": None, "extra": None}
|
||||||
|
|
||||||
script_path = m.group(1)
|
script_path = m.group(1)
|
||||||
name = m.group("name")
|
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()
|
after = exec_start[m.end():].strip()
|
||||||
qnum = None
|
qnum = None
|
||||||
extra = 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}
|
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:
|
def is_unit_active(service_name: str) -> bool:
|
||||||
unit = _systemctl_unit_name(service_name)
|
unit = _systemctl_unit_name(service_name)
|
||||||
p = subprocess.run(["systemctl", "is-active", "--quiet", unit])
|
p = subprocess.run(["systemctl", "is-active", "--quiet", unit])
|
||||||
|
|||||||
Reference in New Issue
Block a user