chore: push all design docs, V2 plan specs, and current working state

Includes AirPlan design documents, AircOding-alpha1-plan, AirPlanV2,
AirPlan-ParaV2, AirPlan-Para V1 reference docs, and all working code
changes across packages.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
AirCoding
2026-06-12 17:12:29 +08:00
parent 8f55c962bb
commit ae44be31d5
364 changed files with 46779 additions and 2812 deletions

View File

@@ -0,0 +1,320 @@
#!/usr/bin/env python3
"""Run local cppcheck scans and maintain AirSDB reports."""
from __future__ import annotations
import argparse
import json
import os
import shlex
import subprocess
import sys
import xml.etree.ElementTree as ET
from collections import Counter
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Dict, List, Optional
from airsdb_mode import ensure_files, read_env, setup_cppcheck, stamp
DEFAULT_ENABLE = "warning,style,performance,portability,information"
DEFAULT_CHECK_LEVEL = "exhaustive"
DEFAULT_EXCLUDES = [
".git",
".airsdb",
"node_modules",
"build",
"cmake-build-debug",
"cmake-build-release",
"dist",
"out",
"vendor",
"third_party",
"external",
]
def iso() -> str:
return datetime.now(timezone.utc).isoformat()
def rel(project: Path, path: str) -> str:
if not path:
return ""
try:
return str(Path(path).resolve().relative_to(project))
except Exception:
return path
def find_project_file(project: Path, configured: str = "") -> Optional[Path]:
candidates = []
if configured:
candidates.append(Path(configured))
candidates.extend(
[
project / "compile_commands.json",
project / "build" / "compile_commands.json",
project / "cmake-build-debug" / "compile_commands.json",
project / "cmake-build-release" / "compile_commands.json",
]
)
for candidate in candidates:
p = candidate if candidate.is_absolute() else project / candidate
if p.exists():
return p.resolve()
return None
def build_args(
project: Path,
tool: str,
project_file: str = "",
target: str = "",
enable: str = DEFAULT_ENABLE,
check_level: str = DEFAULT_CHECK_LEVEL,
jobs: int = 1,
std: str = "",
extra: Optional[List[str]] = None,
) -> List[str]:
build_dir = project / "AirPlan" / "state" / "airsdb" / "cppcheck-build"
build_dir.mkdir(parents=True, exist_ok=True)
args = [
tool,
f"--enable={enable}",
f"--check-level={check_level}",
"--inconclusive",
"--inline-suppr",
"--quiet",
"--xml",
"--xml-version=2",
f"--cppcheck-build-dir={build_dir}",
]
if jobs > 1:
args.append(f"-j{jobs}")
if std:
args.append(f"--std={std}")
for item in extra or []:
if item:
args.extend(shlex.split(item, posix=(os.name != "nt")))
pf = find_project_file(project, project_file)
if pf:
args.append(f"--project={pf}")
else:
for ignored in DEFAULT_EXCLUDES:
args.append(f"-i{ignored}")
args.append(target or ".")
return args
def command_text(args: List[str]) -> str:
if os.name == "nt":
return subprocess.list2cmdline(args)
return shlex.join(args)
def extract_xml(text: str) -> str:
start = text.find("<?xml")
if start >= 0:
return text[start:]
start = text.find("<results")
if start >= 0:
return text[start:]
return text
def parse_cppcheck_xml(xml_text: str, project: Path) -> Dict[str, Any]:
xml_text = extract_xml(xml_text)
findings: List[Dict[str, Any]] = []
counts: Counter[str] = Counter()
if not xml_text.strip():
return {"counts": {}, "findings": [], "parseError": "empty_xml"}
try:
root = ET.fromstring(xml_text)
except ET.ParseError as exc:
return {"counts": {}, "findings": [], "parseError": str(exc)}
for error in root.findall(".//error"):
severity = error.attrib.get("severity", "unknown")
counts[severity] += 1
locations = []
for loc in error.findall("location"):
locations.append(
{
"file": rel(project, loc.attrib.get("file", "")),
"line": loc.attrib.get("line", ""),
"info": loc.attrib.get("info", ""),
}
)
findings.append(
{
"severity": severity,
"id": error.attrib.get("id", ""),
"msg": error.attrib.get("msg", ""),
"cwe": error.attrib.get("cwe", ""),
"locations": locations,
}
)
return {"counts": dict(counts), "findings": findings, "parseError": ""}
def top_findings(findings: List[Dict[str, Any]], limit: int = 12) -> List[str]:
rank = {"error": 0, "warning": 1, "performance": 2, "portability": 3, "style": 4, "information": 5}
def key(item: Dict[str, Any]) -> tuple[int, str]:
return (rank.get(item.get("severity", ""), 9), item.get("id", ""))
lines = []
for item in sorted(findings, key=key)[:limit]:
loc = item.get("locations") or [{}]
first = loc[0]
where = first.get("file", "")
if first.get("line"):
where += f":{first['line']}"
cwe = f" CWE-{item['cwe']}" if item.get("cwe") else ""
lines.append(f"[{item.get('severity')}:{item.get('id')}{cwe}] {where} {item.get('msg')}".strip())
return lines
def append_staticanalysis(project: Path, data: Dict[str, Any]) -> None:
ensure_files(project)
path = project / "AirPlan" / "docs" / "staticanalysis.md"
counts = data.get("counts") or {}
counts_text = ", ".join(f"{k}={v}" for k, v in sorted(counts.items())) or "none"
tops = data.get("topFindings") or []
top_text = "\n".join(f" - {line}" for line in tops) if tops else " - none"
result = "failed" if data.get("returnCode", 0) not in [0, None] else ("findings" if data.get("findingCount", 0) else "ok")
entry = (
f"\n### {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}: AirSDB cppcheck\n\n"
f"- Target: {data.get('target', 'local')}\n"
f"- Tool: `{data.get('tool', '')}` {data.get('version', '')}\n"
f"- Command: `{data.get('command', '')}`\n"
f"- Result: {result}; returnCode={data.get('returnCode', '')}\n"
f"- Counts: {counts_text}\n"
f"- Reports: `{data.get('xmlReport', '')}`, `{data.get('jsonReport', '')}`\n"
f"- Top findings:\n{top_text}\n"
f"- AirDbg/AirDo handoff: {data.get('handoff', 'Review top findings before code-level repair or todo validation.')}\n"
f"- Residual risk: cppcheck is static analysis and may miss runtime, integration, configuration, or dependency issues.\n"
)
path.open("a", encoding="utf-8", newline="\n").write(entry)
def write_reports(project: Path, data: Dict[str, Any], xml_text: str, prefix: str = "cppcheck") -> Dict[str, Any]:
reports = project / "AirPlan" / "state" / "airsdb" / "reports"
reports.mkdir(parents=True, exist_ok=True)
ts = stamp()
xml_path = reports / f"{ts}-{prefix}.xml"
json_path = reports / f"{ts}-{prefix}.json"
xml_path.write_text(xml_text, encoding="utf-8", newline="\n")
data["xmlReport"] = str(xml_path)
data["jsonReport"] = str(json_path)
json_path.write_text(json.dumps(data, ensure_ascii=False, indent=2) + "\n", encoding="utf-8", newline="\n")
return data
def run_scan(args: argparse.Namespace) -> int:
project = Path(args.project).expanduser().resolve()
ensure_files(project)
setup = setup_cppcheck(project, auto=not args.no_auto_configure)
if setup.get("status") != "ok":
print(f"airsdb_scan_status=blocked")
for key, value in setup.items():
print(f"{key}={value}")
return 2
tool = setup["tool"]
env = read_env(project / "AirPlan" / "state" / "airsdb" / "tool.env")
extra = list(args.extra or [])
if env.get("AIRSDB_CPPCHECK_OPTIONS"):
extra.append(env["AIRSDB_CPPCHECK_OPTIONS"])
cmd = build_args(project, tool, args.project_file, args.target, args.enable, args.check_level, args.jobs, args.std, extra)
if args.action == "command":
print("airsdb_command_status=ok")
print(f"command={command_text(cmd)}")
return 0
try:
run = subprocess.run(cmd, cwd=project, capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=args.timeout)
except subprocess.TimeoutExpired as exc:
data = {
"target": "local",
"status": "timeout",
"tool": tool,
"version": setup.get("version", ""),
"checkLevel": args.check_level,
"command": command_text(cmd),
"returnCode": "timeout",
"stdout": exc.stdout or "",
"stderr": exc.stderr or "",
"capturedAt": iso(),
"counts": {},
"findings": [],
"findingCount": 0,
"topFindings": [],
}
data = write_reports(project, data, str(exc.stderr or ""), "cppcheck-timeout")
append_staticanalysis(project, data)
print("airsdb_scan_status=timeout")
print(f"jsonReport={data['jsonReport']}")
return 1
xml_text = extract_xml(run.stderr or "")
parsed = parse_cppcheck_xml(xml_text, project)
findings = parsed.get("findings", [])
data = {
"target": "local",
"status": "ok" if run.returncode == 0 else "failed",
"tool": tool,
"version": setup.get("version", ""),
"checkLevel": args.check_level,
"command": command_text(cmd),
"returnCode": run.returncode,
"stdout": run.stdout.strip(),
"stderrPreview": (run.stderr or "")[:2000],
"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, "cppcheck")
append_staticanalysis(project, data)
print(f"airsdb_scan_status={data['status']}")
print(f"findingCount={data['findingCount']}")
print(f"xmlReport={data['xmlReport']}")
print(f"jsonReport={data['jsonReport']}")
return 0 if run.returncode == 0 else 1
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(description="Run local AirSDB cppcheck analysis.")
parser.add_argument("--project", default=".")
parser.add_argument("--action", choices=["setup", "status", "command", "scan"], default="scan")
parser.add_argument("--project-file", default="")
parser.add_argument("--target", 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("--jobs", type=int, default=1)
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")
return parser.parse_args()
def main() -> None:
args = parse_args()
project = Path(args.project).expanduser().resolve()
if args.action in ["setup", "status"]:
ensure_files(project)
setup = setup_cppcheck(project, auto=(args.action == "setup" and not args.no_auto_configure))
print(f"airsdb_cppcheck_status={setup.get('status', 'unknown')}")
for key, value in setup.items():
if key != "status":
print(f"{key}={value}")
raise SystemExit(0 if setup.get("status") == "ok" else 2)
raise SystemExit(run_scan(args))
if __name__ == "__main__":
main()

View File

@@ -0,0 +1,318 @@
#!/usr/bin/env python3
"""Bootstrap AirSDB static-analysis context and cppcheck tooling."""
from __future__ import annotations
import argparse
import json
import os
import platform
import shutil
import subprocess
from datetime import datetime, timezone
from pathlib import Path
from typing import Dict, Optional, Tuple
MARKER_BEGIN = "<!-- AIRSDB:BEGIN -->"
MARKER_END = "<!-- AIRSDB:END -->"
def now_iso() -> str:
return datetime.now(timezone.utc).isoformat()
def stamp() -> str:
return datetime.now().strftime("%Y%m%d-%H%M%S")
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_CPPCHECK", "AIRSDB_CPPCHECK_OPTIONS"]:
if os.environ.get(key):
vals[key] = os.environ[key]
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_files(project: Path) -> None:
air = project / "AirPlan" / "state" / "airsdb"
air.mkdir(parents=True, exist_ok=True)
example = air / "tool.env.example"
if not example.exists():
example.write_text(
"# AirSDB local cppcheck configuration\n"
"AIRSDB_CPPCHECK=cppcheck\n"
"AIRSDB_CPPCHECK_OPTIONS=\n"
"# AirSDB defaults to --check-level=exhaustive for maximum branch analysis detail.\n",
encoding="utf-8",
newline="\n",
)
gitignore = air / ".gitignore"
old = gitignore.read_text(encoding="utf-8") if gitignore.exists() else ""
for item in ["tool.env", "remote-device.env", "reports/", "cppcheck-build/"]:
if item not in old.splitlines():
old = (old.rstrip() + f"\n{item}\n").lstrip()
gitignore.write_text(old, encoding="utf-8", newline="\n")
static = project / "AirPlan" / "docs" / "staticanalysis.md"
if not static.exists():
static.write_text(
"# Static Analysis\n\n"
"Short AI-context reports for AirSDB cppcheck runs. Keep entries concise.\n\n"
"## Entry Template\n\n"
"### YYYY-MM-DD HH:MM:SS: AirSDB cppcheck\n\n"
"- Target: TODO\n"
"- Tool: TODO\n"
"- Command: TODO\n"
"- Result: TODO\n"
"- Counts: TODO\n"
"- Reports: TODO\n"
"- Top findings: TODO\n"
"- AirDbg/AirDo handoff: TODO\n"
"- Residual risk: TODO\n",
encoding="utf-8",
newline="\n",
)
def airsdb_agents_block() -> str:
return f"""{MARKER_BEGIN}
## AirSDB Static Analysis Workflow
1. Use AirSDB for `/airsdb` sessions that run cppcheck static analysis for C/C++ code quality, security-relevant defects, and debugging evidence.
2. On first entry, detect cppcheck; if missing, auto-configure it with the best available package manager or stop with install instructions.
3. Maintain `AirPlan/docs/staticanalysis.md` as the short AI-context report for AirDbg and AirDo. Keep detailed XML/JSON artifacts under `AirPlan/state/airsdb/reports/`.
4. Prefer `compile_commands.json` when available; otherwise scan the narrowest useful source tree with common generated/vendor directories excluded.
5. Default local and remote cppcheck scans to `--check-level=exhaustive` so AirSDB provides the most detailed branch-analysis evidence it can.
6. For local analysis, use `airsdb_cppcheck.py`; for remote targets over SSH, use `airsdb_remote_device.py` before local analysis.
7. When remote cppcheck is missing, the remote helper may auto-configure it with non-interactive package-manager commands; if sudo/password or unsupported OS blocks it, stop and record the blocker.
8. Record command, target, report paths, counts, top findings, AirDbg/AirDo handoff, and residual risk in `AirPlan/docs/staticanalysis.md`.
9. Update ADR/C4 only when static analysis tooling becomes a durable project boundary or changes implementation decisions.
{MARKER_END}
"""
def upsert_agents_md(path: Path) -> str:
block = airsdb_agents_block().rstrip() + "\n"
if path.exists():
original = path.read_text(encoding="utf-8")
existed = True
else:
original = "# AGENTS.md\n\n"
existed = False
begin = original.find(MARKER_BEGIN)
end = original.find(MARKER_END)
if begin >= 0 and end > begin:
end += len(MARKER_END)
updated = original[:begin].rstrip() + "\n\n" + block + original[end:].lstrip()
status = "updated"
else:
updated = original.rstrip() + "\n\n" + block
status = "updated" if existed else "created"
path.write_text(updated, encoding="utf-8", newline="\n")
return status
def run_args(args: list[str], timeout: int = 600) -> subprocess.CompletedProcess[str]:
return subprocess.run(args, capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=timeout)
def run_shell(cmd: str, timeout: int = 600) -> subprocess.CompletedProcess[str]:
return subprocess.run(cmd, shell=True, capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=timeout)
def cppcheck_version(tool: str) -> str:
try:
result = run_args([tool, "--version"], timeout=20)
except Exception:
return ""
text = (result.stdout or result.stderr).strip()
return text.splitlines()[0] if text else ""
def detect_cppcheck(project: Path) -> Tuple[Optional[str], str]:
env = read_env(project / "AirPlan" / "state" / "airsdb" / "tool.env")
candidates = []
configured = env.get("AIRSDB_CPPCHECK", "").strip()
if configured:
candidates.append(configured)
candidates.append("cppcheck")
if platform.system().lower() == "windows":
candidates.extend(
[
r"C:\Program Files\Cppcheck\cppcheck.exe",
r"C:\Program Files (x86)\Cppcheck\cppcheck.exe",
]
)
for candidate in candidates:
path = shutil.which(candidate) or candidate
if Path(path).exists() or shutil.which(path):
version = cppcheck_version(path)
if version:
return path, version
return None, ""
def install_cppcheck(project: Path) -> Dict[str, str]:
system = platform.system().lower()
attempts = []
if system == "windows":
package_commands = [
["winget", "install", "--id", "Cppcheck.Cppcheck", "-e", "--accept-source-agreements", "--accept-package-agreements", "--silent"],
["choco", "install", "cppcheck", "-y"],
["scoop", "install", "cppcheck"],
]
for cmd in package_commands:
exe = shutil.which(cmd[0])
if not exe:
continue
run = run_args([exe] + cmd[1:], timeout=1200)
attempts.append({"command": " ".join(cmd), "returncode": str(run.returncode)})
tool, version = detect_cppcheck(project)
if tool:
write_env(project / "AirPlan" / "state" / "airsdb" / "tool.env", {"AIRSDB_CPPCHECK": tool})
return {"status": "ok", "tool": tool, "version": version, "installer": cmd[0], "attempts": json.dumps(attempts)}
return {
"status": "blocked",
"reason": "cppcheck_not_found",
"hint": "Install Cppcheck from https://cppcheck.sourceforge.io/ or configure AIRSDB_CPPCHECK.",
"attempts": json.dumps(attempts),
}
commands = [
"if command -v apt-get >/dev/null 2>&1; then sudo -n apt-get update && sudo -n apt-get install -y cppcheck; "
"elif command -v dnf >/dev/null 2>&1; then sudo -n dnf install -y cppcheck; "
"elif command -v yum >/dev/null 2>&1; then sudo -n yum install -y cppcheck; "
"elif command -v apk >/dev/null 2>&1; then sudo -n apk add cppcheck; "
"elif command -v pacman >/dev/null 2>&1; then sudo -n 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 -n port install cppcheck; "
"else exit 42; fi"
]
run = run_shell(commands[0], timeout=1200)
attempts.append({"command": "system-package-manager", "returncode": str(run.returncode)})
tool, version = detect_cppcheck(project)
if tool:
write_env(project / "AirPlan" / "state" / "airsdb" / "tool.env", {"AIRSDB_CPPCHECK": tool})
return {"status": "ok", "tool": tool, "version": version, "installer": "system-package-manager", "attempts": json.dumps(attempts)}
return {
"status": "blocked",
"reason": "cppcheck_not_found",
"hint": "Install cppcheck with the OS package manager or configure AIRSDB_CPPCHECK.",
"attempts": json.dumps(attempts),
}
def setup_cppcheck(project: Path, auto: bool = True) -> Dict[str, str]:
ensure_files(project)
tool, version = detect_cppcheck(project)
if tool:
write_env(project / "AirPlan" / "state" / "airsdb" / "tool.env", {"AIRSDB_CPPCHECK": tool})
return {"status": "ok", "tool": tool, "version": version, "autoConfigure": "skipped"}
if not auto:
return {"status": "blocked", "reason": "cppcheck_not_found", "example": str(project / "AirPlan" / "state" / "airsdb" / "tool.env.example")}
return install_cppcheck(project)
def artifact_map(project: Path) -> Dict[str, Path]:
return {
"AGENTS.md": project / "AirPlan" / "AGENTS.md",
"staticanalysis": project / "AirPlan" / "docs" / "staticanalysis.md",
"state": project / "AirPlan" / "state" / "airsdb" / "state.json",
"tool_env": project / "AirPlan" / "state" / "airsdb" / "tool.env",
}
def write_state(path: Path, enabled: bool, project: Path, tool_info: Dict[str, str]) -> None:
artifacts = artifact_map(project)
payload = {
"enabled": enabled,
"updatedAt": now_iso(),
"projectRoot": str(project),
"cppcheck": tool_info,
"artifactHealth": {name: p.exists() for name, p in artifacts.items() if name != "state"},
}
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8", newline="\n")
def enter_mode(project: Path, auto: bool = True) -> Tuple[str, Dict[str, str]]:
ensure_files(project)
results = {"AGENTS.md": upsert_agents_md(project / "AirPlan" / "AGENTS.md"), "staticanalysis": "exists"}
tool_info = setup_cppcheck(project, auto=auto)
for key, value in tool_info.items():
results[f"cppcheck_{key}"] = value
write_state(project / "AirPlan" / "state" / "airsdb" / "state.json", True, project, tool_info)
return "enabled", results
def exit_mode(project: Path) -> Tuple[str, Dict[str, str]]:
ensure_files(project)
tool_info = setup_cppcheck(project, auto=False)
write_state(project / "AirPlan" / "state" / "airsdb" / "state.json", False, project, tool_info)
return "disabled", {}
def status_mode(project: Path) -> Tuple[str, Dict[str, str]]:
ensure_files(project)
state = project / "AirPlan" / "state" / "airsdb" / "state.json"
enabled = False
if state.exists():
try:
enabled = bool(json.loads(state.read_text(encoding="utf-8")).get("enabled"))
except json.JSONDecodeError:
enabled = False
tool_info = setup_cppcheck(project, auto=False)
results = {name: ("ok" if path.exists() else "missing") for name, path in artifact_map(project).items() if name != "state"}
for key, value in tool_info.items():
results[f"cppcheck_{key}"] = value
return ("enabled" if enabled else "disabled"), results
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(description="Manage AirSDB static-analysis artifacts.")
parser.add_argument("--mode", choices=["enter", "setup", "exit", "status"], default="enter")
parser.add_argument("--project", default=".")
parser.add_argument("--no-auto-configure", action="store_true")
return parser.parse_args()
def main() -> None:
args = parse_args()
project = Path(args.project).expanduser().resolve()
if args.mode in ["enter", "setup"]:
mode_state, result = enter_mode(project, auto=not args.no_auto_configure)
elif args.mode == "exit":
mode_state, result = exit_mode(project)
else:
mode_state, result = status_mode(project)
print(f"airsdb_mode={mode_state}")
print(f"project_root={project}")
for key, value in result.items():
print(f"{key}={value}")
if __name__ == "__main__":
main()

View 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()