166 lines
7.5 KiB
Python
166 lines
7.5 KiB
Python
#!/usr/bin/env python3
|
|
"""Validate safe runner configuration for disposable 1C extension validation."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import re
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
ALLOWED_RUNNER_KINDS = {"manual", "designer_cli", "onescript", "custom"}
|
|
ALLOWED_VALIDATION_MODES = {"manual", "load_and_syntax_check", "load_syntax_and_smoke"}
|
|
FORBIDDEN_BASE_MARKERS = {"prod", "production", "рабоч", "боев", "real", "main"}
|
|
SECRET_KEY_RE = re.compile(r"(password|passwd|pwd|secret|token|ключ|парол)", re.IGNORECASE)
|
|
CONNECTION_SECRET_RE = re.compile(r"(pwd|password|usr|user)\s*=", re.IGNORECASE)
|
|
|
|
|
|
def load_json(path: Path) -> dict[str, Any]:
|
|
return json.loads(path.read_text(encoding="utf-8-sig"))
|
|
|
|
|
|
def write_json(path: Path, data: dict[str, Any]) -> None:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(json.dumps(data, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
|
|
|
|
|
def issue(severity: str, code: str, message: str, *, path: Path | str | None = None, detail: dict[str, Any] | None = None) -> dict[str, Any]:
|
|
result: dict[str, Any] = {"severity": severity, "code": code, "message": message}
|
|
if path is not None:
|
|
result["path"] = str(path)
|
|
if detail:
|
|
result["detail"] = detail
|
|
return result
|
|
|
|
|
|
def find_secret_keys(value: Any, prefix: str = "") -> list[str]:
|
|
hits: list[str] = []
|
|
if isinstance(value, dict):
|
|
for key, nested in value.items():
|
|
current = f"{prefix}.{key}" if prefix else str(key)
|
|
if SECRET_KEY_RE.search(str(key)):
|
|
hits.append(current)
|
|
hits.extend(find_secret_keys(nested, current))
|
|
elif isinstance(value, list):
|
|
for index, nested in enumerate(value):
|
|
hits.extend(find_secret_keys(nested, f"{prefix}[{index}]"))
|
|
return hits
|
|
|
|
|
|
def string_contains_forbidden_marker(value: str) -> str | None:
|
|
lowered = value.casefold()
|
|
for marker in FORBIDDEN_BASE_MARKERS:
|
|
if marker in lowered:
|
|
return marker
|
|
return None
|
|
|
|
|
|
def sanitized_config(config: dict[str, Any]) -> dict[str, Any]:
|
|
allowed = {
|
|
"schema",
|
|
"runner_id",
|
|
"runner_kind",
|
|
"platform_version",
|
|
"platform_bin",
|
|
"disposable_base_ref",
|
|
"disposable_base_kind",
|
|
"disposable_base_confirmed",
|
|
"validation_mode",
|
|
"evidence_root",
|
|
"notes",
|
|
}
|
|
return {key: value for key, value in config.items() if key in allowed}
|
|
|
|
|
|
def check_config(config_path: Path) -> dict[str, Any]:
|
|
findings: list[dict[str, Any]] = []
|
|
if not config_path.exists() or not config_path.is_file():
|
|
findings.append(issue("error", "missing_runner_config", "Runner config file is missing.", path=config_path))
|
|
return build_result(config_path, None, findings)
|
|
|
|
config = load_json(config_path)
|
|
if config.get("schema") != "onec_extension_runner_config.v1":
|
|
findings.append(issue("error", "invalid_runner_config_schema", "Runner config schema must be onec_extension_runner_config.v1.", path=config_path, detail={"schema": config.get("schema")}))
|
|
|
|
runner_id = config.get("runner_id")
|
|
if not isinstance(runner_id, str) or not runner_id.strip():
|
|
findings.append(issue("error", "missing_runner_id", "runner_id is required.", path=config_path))
|
|
|
|
runner_kind = config.get("runner_kind")
|
|
if runner_kind not in ALLOWED_RUNNER_KINDS:
|
|
findings.append(issue("error", "invalid_runner_kind", "runner_kind is not supported.", path=config_path, detail={"allowed": sorted(ALLOWED_RUNNER_KINDS), "actual": runner_kind}))
|
|
|
|
validation_mode = config.get("validation_mode")
|
|
if validation_mode not in ALLOWED_VALIDATION_MODES:
|
|
findings.append(issue("error", "invalid_validation_mode", "validation_mode is not supported.", path=config_path, detail={"allowed": sorted(ALLOWED_VALIDATION_MODES), "actual": validation_mode}))
|
|
|
|
disposable_base_ref = config.get("disposable_base_ref")
|
|
if not isinstance(disposable_base_ref, str) or not disposable_base_ref.strip():
|
|
findings.append(issue("error", "missing_disposable_base_ref", "disposable_base_ref is required.", path=config_path))
|
|
else:
|
|
marker = string_contains_forbidden_marker(disposable_base_ref)
|
|
if marker:
|
|
findings.append(issue("error", "production_like_base_ref", "disposable_base_ref contains a production-like marker.", path=config_path, detail={"marker": marker}))
|
|
if CONNECTION_SECRET_RE.search(disposable_base_ref):
|
|
findings.append(issue("error", "secret_in_disposable_base_ref", "disposable_base_ref must not contain user/password connection data.", path=config_path))
|
|
|
|
if config.get("disposable_base_confirmed") is not True:
|
|
findings.append(issue("error", "disposable_base_not_confirmed", "disposable_base_confirmed must be true.", path=config_path))
|
|
|
|
platform_bin = config.get("platform_bin")
|
|
if platform_bin is not None:
|
|
if not isinstance(platform_bin, str) or not platform_bin.strip():
|
|
findings.append(issue("error", "invalid_platform_bin", "platform_bin must be a non-empty string when provided.", path=config_path))
|
|
elif runner_kind in {"designer_cli", "custom"} and not Path(platform_bin).exists():
|
|
findings.append(issue("warning", "platform_bin_not_found", "platform_bin does not exist on this machine; runner may be remote or not installed here.", path=platform_bin))
|
|
|
|
evidence_root = config.get("evidence_root")
|
|
if evidence_root is not None and (not isinstance(evidence_root, str) or not evidence_root.strip()):
|
|
findings.append(issue("error", "invalid_evidence_root", "evidence_root must be a non-empty string when provided.", path=config_path))
|
|
|
|
secret_keys = find_secret_keys(config)
|
|
for key in secret_keys:
|
|
findings.append(issue("error", "secret_key_in_runner_config", "Runner config must not contain secrets or credentials.", path=config_path, detail={"key": key}))
|
|
|
|
unknown = sorted(set(config) - set(sanitized_config(config)))
|
|
for key in unknown:
|
|
findings.append(issue("warning", "unknown_runner_config_key", "Unknown runner config key will be ignored by the adapter.", path=config_path, detail={"key": key}))
|
|
|
|
return build_result(config_path, sanitized_config(config), findings)
|
|
|
|
|
|
def build_result(config_path: Path, config: dict[str, Any] | None, findings: list[dict[str, Any]]) -> dict[str, Any]:
|
|
errors = [row for row in findings if row.get("severity") == "error"]
|
|
warnings = [row for row in findings if row.get("severity") == "warning"]
|
|
return {
|
|
"schema": "onec_extension_runner_config_check.v1",
|
|
"config_path": str(config_path),
|
|
"config_schema": (config or {}).get("schema"),
|
|
"passed": not errors,
|
|
"sanitized_config": config,
|
|
"findings": findings,
|
|
"counts": {
|
|
"errors": len(errors),
|
|
"warnings": len(warnings),
|
|
},
|
|
}
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description="Validate safe runner configuration for disposable 1C extension validation.")
|
|
parser.add_argument("--config", type=Path, required=True)
|
|
parser.add_argument("--output", type=Path)
|
|
args = parser.parse_args()
|
|
|
|
result = check_config(args.config)
|
|
if args.output:
|
|
write_json(args.output, result)
|
|
print(json.dumps({"output": str(args.output) if args.output else None, "passed": result["passed"], "counts": result["counts"]}, ensure_ascii=False))
|
|
return 0 if result["passed"] else 2
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|