ldr/redmine
4
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Harden portable Redmine CLI runtime and cache writes

Browse Source
This commit is contained in:
Jason Thistlethwaite
2026-10-18 03:18:36 -04:00
parent f6915face2
commit 2e686215ed
19 changed files with 1353 additions and 224 deletions
Download Patch File Download Diff File Expand all files Collapse all files
redmine_ticket_nmi.py
+115 -112
View File
@@ -16,16 +16,21 @@ from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Iterable, Optional
# Prefer env overrides for agents/profiles
DEFAULT_NMI_BIN = os.environ.get(
"NMI_CLI",
"/home/jason/projects/mrscClient-php/bin/nmi",
)
DEFAULT_ENV_FILE = os.environ.get(
"NMI_ENV_FILE",
"/home/jason/projects/mrscClient-php/examples/modern/.env",
)
DEFAULT_MODE = os.environ.get("NMI_MODE", "dev")
# Compatibility constants are environment-independent. Resolve overrides at call
# time so a TUI/worker cannot retain a profile captured during module import.
DEFAULT_NMI_BIN = str(Path(__file__).resolve().parent.parent / "mrscClient-php" / "bin" / "nmi")
DEFAULT_ENV_FILE = None # PHP owns default env-file discovery.
DEFAULT_MODE = "production"
def discover_nmi_bin(value: Optional[str] = None, *, environ=None) -> Path:
env = os.environ if environ is None else environ
selected = value or env.get("NMI_CLI")
if selected:
expanded = str(Path(selected).expanduser())
return Path(shutil.which(expanded, path=env.get("PATH", os.defpath)) or expanded).resolve()
found = shutil.which("nmi", path=env.get("PATH", os.defpath))
return Path(found or DEFAULT_NMI_BIN).resolve()
# Entity types that map cleanly into NMI resolve ids
_TRACKING_TYPES = {
@@ -64,7 +69,7 @@ class NmiRuntimeProfile:
"""Resolved NMI execution settings plus a preflight result."""
mode: str
env_file: Path
env_file: Optional[Path]
nmi_bin: Path
endpoint: Optional[str]
ready: bool
@@ -79,61 +84,65 @@ def resolve_runtime_profile(
endpoint: Optional[str] = None,
environ: Optional[dict[str, str]] = None,
) -> NmiRuntimeProfile:
"""Resolve the exact profile passed to PHP before an NMI subprocess starts.
"""Run the actual PHP resolver offline; never equate file existence to readiness.
An env file that is empty or missing is always rejected. It is too easy for
the checked-in example `.env` to otherwise select a surprising dev/default
path while making the TUI look successful.
With no explicit env file, let NMI perform its own documented discovery.
Environment-only credentials are valid. No service requests are performed.
"""
env = os.environ if environ is None else environ
env = dict(os.environ if environ is None else environ)
requested_mode = (mode or env.get("NMI_MODE") or DEFAULT_MODE).strip().lower()
normalized_mode = "production" if requested_mode in {"prod", "production"} else requested_mode
selected_env_file = Path(env_file or env.get("NMI_ENV_FILE") or DEFAULT_ENV_FILE).expanduser()
selected_nmi_bin = Path(nmi_bin or env.get("NMI_CLI") or DEFAULT_NMI_BIN).expanduser()
normalized_mode = "production" if requested_mode == "prod" else requested_mode
requested_file = env_file or env.get("NMI_ENV_FILE")
selected_env_file = Path(requested_file).expanduser().resolve() if requested_file else None
selected_nmi_bin = discover_nmi_bin(nmi_bin, environ=env)
mode_upper = normalized_mode.upper()
selected_endpoint = (
endpoint
or env.get(f"NMI_ENDPOINT_{mode_upper}")
or env.get("NMI_ENDPOINT")
or None
)
def failed(problem):
return NmiRuntimeProfile(normalized_mode, selected_env_file, selected_nmi_bin, endpoint, False, problem)
if not selected_env_file.is_file():
return NmiRuntimeProfile(
mode=normalized_mode,
env_file=selected_env_file,
nmi_bin=selected_nmi_bin,
endpoint=selected_endpoint,
ready=False,
problem=f"NMI env file does not exist: {selected_env_file}",
)
if selected_env_file.stat().st_size == 0:
return NmiRuntimeProfile(
mode=normalized_mode,
env_file=selected_env_file,
nmi_bin=selected_nmi_bin,
endpoint=selected_endpoint,
ready=False,
problem=f"NMI env file is empty: {selected_env_file}",
)
if normalized_mode not in {"production", "dev"}:
return failed("NMI_MODE must be production or dev.")
if selected_env_file is not None:
try:
if not selected_env_file.is_file():
return failed(f"NMI env file does not exist: {selected_env_file}")
if selected_env_file.stat().st_size == 0:
return failed(f"NMI env file is empty: {selected_env_file}")
except OSError as exc:
return failed(f"Cannot inspect NMI env file: {exc}")
if not selected_nmi_bin.is_file():
return NmiRuntimeProfile(
mode=normalized_mode,
env_file=selected_env_file,
nmi_bin=selected_nmi_bin,
endpoint=selected_endpoint,
ready=False,
problem=f"NMI CLI does not exist: {selected_nmi_bin}",
)
return failed(f"NMI CLI does not exist: {selected_nmi_bin}; set NMI_CLI.")
php = shutil.which("php", path=env.get("PATH", os.defpath))
if not php:
return failed("PHP is not on PATH; NMI requires the php CLI.")
cmd = [php, str(selected_nmi_bin), "--doctor", "--json"]
# Preserve NMI env-file mode when no caller/environment mode was supplied.
if mode or env.get("NMI_MODE"):
cmd.append(f"--mode={normalized_mode}")
if selected_env_file:
cmd.append(f"--env-file={selected_env_file}")
if endpoint:
cmd.append(f"--endpoint={endpoint}")
try:
proc = subprocess.run(cmd, capture_output=True, text=True, timeout=15, check=False, env=env)
except subprocess.TimeoutExpired:
return failed("NMI offline doctor timed out after 15 seconds.")
except OSError as exc:
return failed(f"Could not start NMI doctor: {exc}")
try:
data = json.loads(proc.stdout)
if not isinstance(data, dict):
raise ValueError("expected an object")
except (ValueError, TypeError):
return failed(f"NMI doctor returned invalid diagnostic JSON (exit {proc.returncode}); update NMI_CLI to a compatible bin/nmi.")
ready = proc.returncode == 0 and data.get("ready") is True
resolved_file = data.get("env_file")
return NmiRuntimeProfile(
mode=normalized_mode,
env_file=selected_env_file,
mode=data.get("mode") or normalized_mode,
env_file=Path(resolved_file) if resolved_file else selected_env_file,
nmi_bin=selected_nmi_bin,
endpoint=selected_endpoint,
ready=True,
endpoint=data.get("endpoint") or endpoint,
ready=ready,
problem=None if ready else (data.get("problem") or f"NMI doctor failed (exit {proc.returncode})."),
)
@@ -223,30 +232,35 @@ def _php_bin() -> str:
def run_nmi(
args: list[str],
*,
mode: str = DEFAULT_MODE,
env_file: str = DEFAULT_ENV_FILE,
nmi_bin: str = DEFAULT_NMI_BIN,
mode: Optional[str] = None,
env_file: Optional[str] = None,
nmi_bin: Optional[str] = None,
endpoint: Optional[str] = None,
timeout: int = 60,
) -> subprocess.CompletedProcess[str]:
cmd = [_php_bin(), nmi_bin, *args, f"--mode={mode}", f"--env-file={env_file}"]
cmd = [_php_bin(), str(discover_nmi_bin(nmi_bin)), *args]
selected_mode = mode or os.environ.get("NMI_MODE")
selected_file = env_file or os.environ.get("NMI_ENV_FILE")
if selected_mode:
cmd.append(f"--mode={selected_mode}")
if selected_file:
cmd.append(f"--env-file={Path(selected_file).expanduser().resolve()}")
if endpoint:
cmd.append(f"--endpoint={endpoint}")
return subprocess.run(
cmd,
capture_output=True,
text=True,
timeout=timeout,
check=False,
)
try:
return subprocess.run(cmd, capture_output=True, text=True, timeout=timeout, check=False)
except subprocess.TimeoutExpired:
return subprocess.CompletedProcess(cmd, 124, "", f"NMI command timed out after {timeout} seconds.")
except OSError as exc:
return subprocess.CompletedProcess(cmd, 127, "", f"Could not start NMI CLI: {exc}")
def try_resolve_ids(
plan: NmiLookupPlan,
*,
mode: str = DEFAULT_MODE,
env_file: str = DEFAULT_ENV_FILE,
nmi_bin: str = DEFAULT_NMI_BIN,
mode: Optional[str] = None,
env_file: Optional[str] = None,
nmi_bin: Optional[str] = None,
endpoint: Optional[str] = None,
include_pictures: bool = False,
max_attempts: int = 6,
@@ -294,9 +308,9 @@ def try_resolve_ids(
def try_cs_brief(
plan: NmiLookupPlan,
*,
mode: str = DEFAULT_MODE,
env_file: str = DEFAULT_ENV_FILE,
nmi_bin: str = DEFAULT_NMI_BIN,
mode: Optional[str] = None,
env_file: Optional[str] = None,
nmi_bin: Optional[str] = None,
endpoint: Optional[str] = None,
max_attempts: int = 6,
) -> dict[str, Any]:
@@ -336,9 +350,9 @@ def brief_for_issue(
issue: dict[str, Any],
entities: list[dict[str, str]],
*,
mode: str = DEFAULT_MODE,
env_file: str = DEFAULT_ENV_FILE,
nmi_bin: str = DEFAULT_NMI_BIN,
mode: Optional[str] = None,
env_file: Optional[str] = None,
nmi_bin: Optional[str] = None,
endpoint: Optional[str] = None,
with_history: bool = True,
) -> dict[str, Any]:
@@ -350,7 +364,7 @@ def brief_for_issue(
)
runtime = {
"mode": profile.mode,
"env_file": str(profile.env_file),
"env_file": str(profile.env_file) if profile.env_file else None,
"nmi_bin": str(profile.nmi_bin),
"endpoint": profile.endpoint,
"ready": profile.ready,
@@ -400,6 +414,11 @@ def brief_for_issue(
out["error"] = "no_nmi_candidates"
return out
# Execute exactly the profile checked above, not import-time/caller defaults.
mode = profile.mode
env_file = str(profile.env_file) if profile.env_file else None
nmi_bin = str(profile.nmi_bin)
endpoint = profile.endpoint
brief = try_cs_brief(plan, mode=mode, env_file=env_file, nmi_bin=nmi_bin, endpoint=endpoint)
out["brief"] = brief
out["ok"] = bool(brief.get("ok"))
@@ -504,33 +523,20 @@ def generate_nmi_links(entities: list[dict[str, str]]) -> list[str]:
def try_item_history(
item_no: str,
*,
mode: str = DEFAULT_MODE,
env_file: str = DEFAULT_ENV_FILE,
nmi_bin: str = DEFAULT_NMI_BIN,
mode: Optional[str] = None,
env_file: Optional[str] = None,
nmi_bin: Optional[str] = None,
endpoint: Optional[str] = None,
limit: int = 50,
snapshot: bool = True,
) -> dict[str, Any]:
"""Call nmi item history for a single item_no. Returns structured result."""
cmd = [
"php",
nmi_bin,
"item",
"history",
f"--item-no={item_no}",
f"--limit={limit}",
]
args = ["item", "history", f"--item-no={item_no}", f"--limit={limit}"]
if snapshot:
cmd.append("--snapshot")
if mode:
cmd.extend(["--mode", mode])
if env_file:
cmd.extend(["--env-file", env_file])
if endpoint:
cmd.extend(["--endpoint", endpoint])
args.append("--snapshot")
try:
res = subprocess.run(cmd, capture_output=True, text=True, timeout=30)
res = run_nmi(args, mode=mode, env_file=env_file, nmi_bin=nmi_bin, endpoint=endpoint, timeout=30)
cmd = res.args
if res.returncode != 0:
return {"ok": False, "error": res.stderr.strip() or res.stdout.strip(), "cmd": cmd}
# Try JSON first
@@ -546,9 +552,9 @@ def try_item_history(
def enrich_with_history(
payload: dict[str, Any],
*,
mode: str = DEFAULT_MODE,
env_file: str = DEFAULT_ENV_FILE,
nmi_bin: str = DEFAULT_NMI_BIN,
mode: Optional[str] = None,
env_file: Optional[str] = None,
nmi_bin: Optional[str] = None,
endpoint: Optional[str] = None,
) -> dict[str, Any]:
"""If an item_no is present in the plan, fetch its history and attach."""
@@ -557,7 +563,7 @@ def enrich_with_history(
if not item_no:
return payload
hist = try_item_history(
str(item_no),
str(item_no[0] if isinstance(item_no, list) else item_no),
mode=mode,
env_file=env_file,
nmi_bin=nmi_bin,
@@ -574,9 +580,9 @@ def try_auto_link_redmine_contact(
tracking: list[str] | None = None,
order_id: list[str] | None = None,
item_no: list[str] | None = None,
mode: str = DEFAULT_MODE,
env_file: str = DEFAULT_ENV_FILE,
nmi_bin: str = DEFAULT_NMI_BIN,
mode: Optional[str] = None,
env_file: Optional[str] = None,
nmi_bin: Optional[str] = None,
) -> dict[str, Any]:
"""If we have a Redmine contact_id and identifiers from a ticket, try to resolve in NMI
and automatically create the redmine_contact_links entry.
@@ -602,13 +608,10 @@ def try_auto_link_redmine_contact(
if not ids:
return {"ok": False, "reason": "no_identifiers"}
cmd = [nmi_bin, "link-contact", "--redmine-contact-id", str(contact_id), "--mode", mode]
if env_file and env_file != DEFAULT_ENV_FILE:
cmd.extend(["--env-file", env_file])
cmd.extend(ids)
args = ["link-contact", "--redmine-contact-id", str(contact_id), *ids]
try:
proc = subprocess.run(cmd, capture_output=True, text=True, timeout=90)
proc = run_nmi(args, mode=mode, env_file=env_file, nmi_bin=nmi_bin, timeout=90)
cmd = proc.args
ok = proc.returncode == 0
result = {
"ok": ok,