Initial release: airsdb
This commit is contained in:
admin
314
scripts/airsdb_remote_device.py
Normal file
314
scripts/airsdb_remote_device.py
Normal file
@@ -0,0 +1,314 @@
|
||||
#!/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()
|
||||
Reference in New Issue
Block a user