315 lines
13 KiB
Python
315 lines
13 KiB
Python
#!/usr/bin/env python3
|
|
"""Run AirSDB cppcheck on a remote SSH target."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import base64
|
|
import json
|
|
import os
|
|
import shlex
|
|
import shutil
|
|
import subprocess
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Dict, List
|
|
|
|
from airsdb_cppcheck import append_staticanalysis, parse_cppcheck_xml, top_findings, write_reports
|
|
from airsdb_mode import ensure_files, stamp
|
|
|
|
DEFAULT_ENABLE = "warning,style,performance,portability,information"
|
|
DEFAULT_CHECK_LEVEL = "exhaustive"
|
|
|
|
|
|
def iso() -> str:
|
|
return datetime.now(timezone.utc).isoformat()
|
|
|
|
|
|
def q(value: str) -> str:
|
|
return shlex.quote(str(value))
|
|
|
|
|
|
def split(value: str) -> List[str]:
|
|
return shlex.split(value, posix=True) if value.strip() else []
|
|
|
|
|
|
def read_env(path: Path) -> Dict[str, str]:
|
|
vals: Dict[str, str] = {}
|
|
if path.exists():
|
|
for raw in path.read_text(encoding="utf-8-sig").splitlines():
|
|
line = raw.strip()
|
|
if line and not line.startswith("#") and "=" in line:
|
|
key, value = line.split("=", 1)
|
|
vals[key.strip()] = value.strip().strip('"').strip("'")
|
|
for key in [
|
|
"AIRSDB_REMOTE_SSH_TARGET",
|
|
"AIRSDB_REMOTE_SSH_PORT",
|
|
"AIRSDB_REMOTE_SSH_OPTIONS",
|
|
"AIRSDB_REMOTE_WORKDIR",
|
|
"AIRSDB_REMOTE_PROJECT",
|
|
"AIRSDB_REMOTE_CPPCHECK",
|
|
]:
|
|
if os.environ.get(key):
|
|
vals[key] = os.environ[key]
|
|
vals.setdefault("AIRSDB_REMOTE_SSH_PORT", "22")
|
|
vals.setdefault("AIRSDB_REMOTE_CPPCHECK", "auto")
|
|
return vals
|
|
|
|
|
|
def write_env(path: Path, updates: Dict[str, str]) -> None:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
old = path.read_text(encoding="utf-8") if path.exists() else ""
|
|
keys = set(updates)
|
|
lines = []
|
|
for raw in old.splitlines():
|
|
key = raw.split("=", 1)[0].strip() if "=" in raw and not raw.strip().startswith("#") else None
|
|
if key not in keys:
|
|
lines.append(raw)
|
|
if lines and lines[-1].strip():
|
|
lines.append("")
|
|
for key, value in updates.items():
|
|
if value:
|
|
lines.append(f"{key}={value}")
|
|
path.write_text("\n".join(lines).rstrip() + "\n", encoding="utf-8", newline="\n")
|
|
|
|
|
|
def ensure_remote_files(project: Path) -> None:
|
|
ensure_files(project)
|
|
air = project / "AirPlan" / "state" / "airsdb"
|
|
example = air / "remote-device.env.example"
|
|
if not example.exists():
|
|
example.write_text(
|
|
"# AirSDB remote cppcheck configuration\n"
|
|
"AIRSDB_REMOTE_SSH_TARGET=user@host\n"
|
|
"AIRSDB_REMOTE_SSH_PORT=22\n"
|
|
"AIRSDB_REMOTE_SSH_OPTIONS=\n"
|
|
"AIRSDB_REMOTE_WORKDIR=\n"
|
|
"AIRSDB_REMOTE_PROJECT=/path/to/remote/project\n"
|
|
"AIRSDB_REMOTE_CPPCHECK=auto\n",
|
|
encoding="utf-8",
|
|
newline="\n",
|
|
)
|
|
|
|
|
|
def ssh_args(env: Dict[str, str], cmd: str) -> List[str]:
|
|
target = env.get("AIRSDB_REMOTE_SSH_TARGET", "")
|
|
if not target:
|
|
raise SystemExit("AIRSDB_REMOTE_SSH_TARGET is required")
|
|
args = [shutil.which("ssh") or "ssh"]
|
|
port = env.get("AIRSDB_REMOTE_SSH_PORT", "22")
|
|
if port:
|
|
args += ["-p", port]
|
|
args += split(env.get("AIRSDB_REMOTE_SSH_OPTIONS", ""))
|
|
return args + [target, cmd]
|
|
|
|
|
|
def run(env: Dict[str, str], cmd: str, timeout: int = 60) -> subprocess.CompletedProcess[str]:
|
|
return subprocess.run(ssh_args(env, cmd), capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=timeout)
|
|
|
|
|
|
def remote_home(env: Dict[str, str]) -> str:
|
|
result = run(env, 'printf %s "$HOME"', 15)
|
|
return result.stdout.strip() if result.returncode == 0 else ""
|
|
|
|
|
|
def workdir(env: Dict[str, str]) -> str:
|
|
if env.get("AIRSDB_REMOTE_WORKDIR"):
|
|
return env["AIRSDB_REMOTE_WORKDIR"]
|
|
home = remote_home(env)
|
|
return (home.rstrip("/") + "/.airsdb") if home else ".airsdb"
|
|
|
|
|
|
def detect_tool(env: Dict[str, str]) -> str:
|
|
tool = env.get("AIRSDB_REMOTE_CPPCHECK", "auto").strip()
|
|
if tool and tool != "auto":
|
|
result = run(env, f"command -v {q(tool)} >/dev/null 2>&1 && command -v {q(tool)} || test -x {q(tool)} && printf %s {q(tool)}", 20)
|
|
return result.stdout.strip().splitlines()[-1] if result.returncode == 0 and result.stdout.strip() else ""
|
|
result = run(env, 'command -v cppcheck >/dev/null 2>&1 && command -v cppcheck', 20)
|
|
return result.stdout.strip().splitlines()[-1] if result.returncode == 0 and result.stdout.strip() else ""
|
|
|
|
|
|
def remote_version(env: Dict[str, str], tool: str) -> str:
|
|
result = run(env, f"{q(tool)} --version", 20)
|
|
text = (result.stdout or result.stderr).strip()
|
|
return text.splitlines()[0] if result.returncode == 0 and text else ""
|
|
|
|
|
|
def install_tool(env: Dict[str, str]) -> subprocess.CompletedProcess[str]:
|
|
script = """set -e
|
|
if command -v cppcheck >/dev/null 2>&1; then exit 0; fi
|
|
if [ "$(id -u 2>/dev/null || echo 1)" = "0" ]; then SUDO=""; else SUDO="sudo -n"; fi
|
|
if command -v apt-get >/dev/null 2>&1; then $SUDO apt-get update && $SUDO apt-get install -y cppcheck;
|
|
elif command -v dnf >/dev/null 2>&1; then $SUDO dnf install -y cppcheck;
|
|
elif command -v yum >/dev/null 2>&1; then $SUDO yum install -y cppcheck;
|
|
elif command -v apk >/dev/null 2>&1; then $SUDO apk add cppcheck;
|
|
elif command -v pacman >/dev/null 2>&1; then $SUDO pacman -Sy --noconfirm cppcheck;
|
|
elif command -v brew >/dev/null 2>&1; then brew install cppcheck;
|
|
elif command -v port >/dev/null 2>&1; then $SUDO port install cppcheck;
|
|
else exit 42; fi"""
|
|
return run(env, script, 1200)
|
|
|
|
|
|
def setup(project: Path, env: Dict[str, str], auto: bool = True) -> Dict[str, str]:
|
|
if not env.get("AIRSDB_REMOTE_SSH_TARGET"):
|
|
return {"status": "blocked", "reason": "missing_remote_target", "example": str(project / "AirPlan" / "state" / "airsdb" / "remote-device.env.example")}
|
|
if not shutil.which("ssh"):
|
|
return {"status": "blocked", "reason": "ssh_not_found"}
|
|
probe = run(env, "printf ok", 20)
|
|
if probe.returncode:
|
|
return {"status": "blocked", "reason": "ssh_probe_failed", "stderr": probe.stderr.strip()}
|
|
wd = workdir(env)
|
|
mk = run(env, f"mkdir -p {q(wd)} {q(wd.rstrip('/') + '/reports')} {q(wd.rstrip('/') + '/cppcheck-build')}", 30)
|
|
if mk.returncode:
|
|
return {"status": "blocked", "reason": "remote_workdir_failed", "stderr": mk.stderr.strip()}
|
|
tool = detect_tool(env)
|
|
install = "skipped"
|
|
if not tool and auto:
|
|
inst = install_tool(env)
|
|
install = "ok" if inst.returncode == 0 else f"failed:{inst.returncode}"
|
|
tool = detect_tool(env)
|
|
updates = {"AIRSDB_REMOTE_WORKDIR": wd}
|
|
if tool:
|
|
updates["AIRSDB_REMOTE_CPPCHECK"] = tool
|
|
write_env(project / "AirPlan" / "state" / "airsdb" / "remote-device.env", updates)
|
|
version = remote_version(env, tool) if tool else ""
|
|
return {
|
|
"status": "ok" if tool else "blocked",
|
|
"target": env.get("AIRSDB_REMOTE_SSH_TARGET", ""),
|
|
"workdir": wd,
|
|
"tool": tool,
|
|
"version": version,
|
|
"autoConfigure": install,
|
|
"hint": "" if tool else "Install cppcheck on the remote target or set AIRSDB_REMOTE_CPPCHECK.",
|
|
}
|
|
|
|
|
|
def scan_cmd(
|
|
env: Dict[str, str],
|
|
tool: str,
|
|
remote_project: str,
|
|
remote_xml: str,
|
|
enable: str,
|
|
check_level: str,
|
|
std: str,
|
|
extra: List[str],
|
|
) -> str:
|
|
build_dir = workdir(env).rstrip("/") + "/cppcheck-build"
|
|
report_dir = remote_xml.rsplit("/", 1)[0] if "/" in remote_xml else "."
|
|
options = [
|
|
q(tool),
|
|
f"--enable={q(enable)}",
|
|
f"--check-level={q(check_level)}",
|
|
"--inconclusive",
|
|
"--inline-suppr",
|
|
"--quiet",
|
|
"--xml",
|
|
"--xml-version=2",
|
|
f"--cppcheck-build-dir={q(build_dir)}",
|
|
]
|
|
if std:
|
|
options.append(f"--std={q(std)}")
|
|
for item in extra:
|
|
if item:
|
|
options.append(item)
|
|
return (
|
|
f"cd {q(remote_project)} && mkdir -p {q(report_dir)} {q(build_dir)} && "
|
|
"if [ -f compile_commands.json ]; then AIRSDB_PROJECT_ARG='--project=compile_commands.json'; "
|
|
"elif [ -f build/compile_commands.json ]; then AIRSDB_PROJECT_ARG='--project=build/compile_commands.json'; "
|
|
"elif [ -f cmake-build-debug/compile_commands.json ]; then AIRSDB_PROJECT_ARG='--project=cmake-build-debug/compile_commands.json'; "
|
|
"elif [ -f cmake-build-release/compile_commands.json ]; then AIRSDB_PROJECT_ARG='--project=cmake-build-release/compile_commands.json'; "
|
|
"else AIRSDB_PROJECT_ARG='.'; fi; "
|
|
+ " ".join(options)
|
|
+ f" $AIRSDB_PROJECT_ARG 2> {q(remote_xml)}; AIRSDB_RC=$?; "
|
|
'printf "airsdb_cppcheck_rc=%s\\n" "$AIRSDB_RC"; '
|
|
'printf "airsdb_project_arg=%s\\n" "$AIRSDB_PROJECT_ARG"; '
|
|
"exit 0"
|
|
)
|
|
|
|
|
|
def emit(prefix: str, data: Dict[str, object]) -> None:
|
|
print(f"{prefix}_status={data.get('status', 'unknown')}")
|
|
for key, value in data.items():
|
|
if key != "status":
|
|
print(f"{key}={json.dumps(value, ensure_ascii=False) if isinstance(value, (dict, list)) else value}")
|
|
|
|
|
|
def parse_stdout_value(stdout: str, key: str) -> str:
|
|
prefix = f"{key}="
|
|
for line in stdout.splitlines():
|
|
if line.startswith(prefix):
|
|
return line[len(prefix) :].strip()
|
|
return ""
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser(description="Run AirSDB cppcheck on a remote SSH target.")
|
|
parser.add_argument("--project", default=".")
|
|
parser.add_argument("--action", choices=["setup", "status", "command", "scan"], default="setup")
|
|
parser.add_argument("--remote-project", default="")
|
|
parser.add_argument("--enable", default=DEFAULT_ENABLE)
|
|
parser.add_argument("--check-level", default=DEFAULT_CHECK_LEVEL)
|
|
parser.add_argument("--std", default="")
|
|
parser.add_argument("--timeout", type=int, default=900)
|
|
parser.add_argument("--limit", type=int, default=12)
|
|
parser.add_argument("--extra", action="append", default=[])
|
|
parser.add_argument("--no-auto-configure", action="store_true")
|
|
args = parser.parse_args()
|
|
|
|
project = Path(args.project).expanduser().resolve()
|
|
ensure_remote_files(project)
|
|
env = read_env(project / "AirPlan" / "state" / "airsdb" / "remote-device.env")
|
|
info = setup(project, env, auto=(args.action != "status" and not args.no_auto_configure))
|
|
if args.action in ["setup", "status"] or info.get("status") != "ok":
|
|
emit("airsdb_remote", info)
|
|
raise SystemExit(0 if info.get("status") == "ok" else 2)
|
|
|
|
remote_project = args.remote_project or env.get("AIRSDB_REMOTE_PROJECT", "")
|
|
if not remote_project:
|
|
emit("airsdb_remote", {"status": "blocked", "reason": "missing_remote_project", "example": str(project / "AirPlan" / "state" / "airsdb" / "remote-device.env.example")})
|
|
raise SystemExit(2)
|
|
|
|
remote_xml = info["workdir"].rstrip("/") + "/reports/" + stamp() + "-airsdb-remote-cppcheck.xml"
|
|
cmd = scan_cmd(env, str(info["tool"]), remote_project, remote_xml, args.enable, args.check_level, args.std, args.extra)
|
|
full_command = shlex.join(ssh_args(env, cmd))
|
|
if args.action == "command":
|
|
emit("airsdb_remote", {"status": "ok", "target": info["target"], "remoteProject": remote_project, "remoteXml": remote_xml, "command": full_command})
|
|
return
|
|
|
|
run_result = run(env, cmd, args.timeout)
|
|
rc_text = parse_stdout_value(run_result.stdout, "airsdb_cppcheck_rc")
|
|
cppcheck_rc = int(rc_text) if rc_text.isdigit() else run_result.returncode
|
|
b64 = run(env, f"base64 < {q(remote_xml)}", 120)
|
|
xml_text = ""
|
|
if b64.returncode == 0 and b64.stdout.strip():
|
|
xml_text = base64.b64decode("".join(b64.stdout.split())).decode("utf-8", errors="replace")
|
|
parsed = parse_cppcheck_xml(xml_text, project)
|
|
findings = parsed.get("findings", [])
|
|
data = {
|
|
"target": f"remote:{info['target']}",
|
|
"remoteProject": remote_project,
|
|
"status": "ok" if run_result.returncode == 0 else "failed",
|
|
"tool": info.get("tool", ""),
|
|
"version": info.get("version", ""),
|
|
"checkLevel": args.check_level,
|
|
"command": full_command,
|
|
"returnCode": cppcheck_rc,
|
|
"sshReturnCode": run_result.returncode,
|
|
"stdout": run_result.stdout.strip(),
|
|
"stderr": run_result.stderr.strip(),
|
|
"remoteXml": remote_xml,
|
|
"capturedAt": iso(),
|
|
"counts": parsed.get("counts", {}),
|
|
"findings": findings,
|
|
"findingCount": len(findings),
|
|
"topFindings": top_findings(findings, args.limit),
|
|
"parseError": parsed.get("parseError", ""),
|
|
}
|
|
data = write_reports(project, data, xml_text, "remote-cppcheck")
|
|
append_staticanalysis(project, data)
|
|
emit("airsdb_remote", data)
|
|
raise SystemExit(0 if run_result.returncode == 0 else 1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|