llm/scripts/check_1c_metadata_kind_fixtures.py

#!/usr/bin/env python3
"""Validate and optionally probe rare 1C metadata-kind fixtures."""

from __future__ import annotations

import argparse
import json
import os
import re
import urllib.error
import urllib.request
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Callable


SCHEMA = "onec_metadata_kind_fixture_manifest.v1"
REPORT_SCHEMA = "onec_metadata_kind_fixture_check.v1"
SUPPORTED_MODES = {"extension_saved_state", "dedicated_base_active"}
REFERENCE_USAGE = "external_reference_only_not_vendored"
DESIGNER_ACCESS_MODE = "operating_system_integrated_only"
SECRET_KEY_RE = re.compile(r"(password|passwd|pwd|secret|token|парол|секрет)", re.IGNORECASE)
Rpc = Callable[[str, str, dict[str, Any], float, str], dict[str, Any]]


def load_json(path: Path) -> dict[str, Any]:
    value = json.loads(path.read_text(encoding="utf-8-sig"))
    if not isinstance(value, dict):
        raise ValueError("manifest root must be a JSON object")
    return value


def write_json(path: Path, value: dict[str, Any]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    temporary = path.with_name(path.name + ".tmp")
    temporary.write_text(json.dumps(value, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
    temporary.replace(path)


def rpc(adapter_url: str, method: str, payload: dict[str, Any], timeout: float, service_token: str) -> dict[str, Any]:
    headers = {"Content-Type": "application/json; charset=utf-8"}
    if service_token:
        headers["Authorization"] = f"Bearer {service_token}"
    request = urllib.request.Request(
        adapter_url.rstrip("/") + "/rpc",
        data=json.dumps({"method": method, "payload": payload}, ensure_ascii=False).encode("utf-8"),
        headers=headers,
        method="POST",
    )
    with urllib.request.urlopen(request, timeout=timeout) as response:
        value = json.loads(response.read().decode("utf-8"))
    return value if isinstance(value, dict) else {"status": "error", "error": "response_not_object"}


def finding(code: str, message: str, *, fixture_id: str | None = None) -> dict[str, Any]:
    value = {"severity": "error", "code": code, "message": message}
    if fixture_id:
        value["fixture_id"] = fixture_id
    return value


def secret_key_paths(value: Any, prefix: str = "") -> list[str]:
    hits: list[str] = []
    if isinstance(value, dict):
        for key, nested in value.items():
            path = f"{prefix}.{key}" if prefix else str(key)
            if SECRET_KEY_RE.search(str(key)):
                hits.append(path)
            hits.extend(secret_key_paths(nested, path))
    elif isinstance(value, list):
        for index, nested in enumerate(value):
            hits.extend(secret_key_paths(nested, f"{prefix}[{index}]"))
    return hits


def validate_manifest(manifest: dict[str, Any]) -> list[dict[str, Any]]:
    findings: list[dict[str, Any]] = []
    if manifest.get("schema") != SCHEMA:
        findings.append(finding("invalid_schema", f"schema must be {SCHEMA}"))

    policy = manifest.get("policy")
    if not isinstance(policy, dict):
        findings.append(finding("missing_policy", "policy must be an object"))
    else:
        if policy.get("provisioning_tool") != "1c_designer":
            findings.append(finding("unsafe_provisioning_tool", "fixtures must be provisioned by 1C Designer"))
        if policy.get("direct_sql_write_allowed") is not False:
            findings.append(finding("direct_sql_write_not_forbidden", "direct SQL fixture writes must be forbidden"))
        if policy.get("credentials_in_manifest_allowed") is not False:
            findings.append(finding("credentials_not_forbidden", "credentials must be forbidden in the manifest"))

    for path in secret_key_paths(manifest):
        findings.append(finding("credential_key_in_manifest", f"credential-like key is forbidden: {path}"))

    designer = manifest.get("designer")
    if not isinstance(designer, dict):
        findings.append(finding("missing_designer_requirements", "designer requirements must be an object"))
        designer = {}
    required_version = str(designer.get("required_version") or "")
    if not re.fullmatch(r"\d+\.\d+\.\d+\.\d+", required_version):
        findings.append(finding("invalid_designer_version", "designer.required_version must be an exact four-part version"))
    if designer.get("access_mode") != DESIGNER_ACCESS_MODE:
        findings.append(finding("unsafe_designer_access_mode", f"designer.access_mode must be {DESIGNER_ACCESS_MODE}"))
    if designer.get("credential_arguments_allowed") is not False:
        findings.append(finding("designer_credential_arguments_not_forbidden", "Designer credential arguments must be forbidden"))
    export_script = str(designer.get("export_script") or "")
    if not export_script.endswith(".ps1") or Path(export_script).is_absolute():
        findings.append(finding("invalid_designer_export_script", "designer.export_script must be a repository-relative PowerShell path"))

    external_reference = manifest.get("external_reference")
    if not isinstance(external_reference, dict):
        findings.append(finding("missing_external_reference", "external_reference must be an object"))
        external_reference = {}
    if external_reference.get("usage") != REFERENCE_USAGE:
        findings.append(finding("unsafe_external_reference_usage", f"external_reference.usage must be {REFERENCE_USAGE}"))
    if not str(external_reference.get("repository") or "").startswith("https://"):
        findings.append(finding("invalid_external_reference_repository", "external_reference.repository must be an HTTPS URL"))
    if not re.fullmatch(r"[0-9a-f]{40}", str(external_reference.get("commit") or "")):
        findings.append(finding("invalid_external_reference_commit", "external_reference.commit must be a full Git commit"))
    for key in ("source_path", "license"):
        if not str(external_reference.get(key) or "").strip():
            findings.append(finding("incomplete_external_reference", f"external_reference.{key} is required"))

    required_kinds = manifest.get("required_kinds")
    if not isinstance(required_kinds, list) or not required_kinds or any(not isinstance(item, str) or not item for item in required_kinds):
        findings.append(finding("invalid_required_kinds", "required_kinds must be a non-empty string array"))
        required_kinds = []

    fixtures = manifest.get("fixtures")
    if not isinstance(fixtures, list):
        findings.append(finding("invalid_fixtures", "fixtures must be an array"))
        return findings

    seen_ids: set[str] = set()
    seen_kinds: set[str] = set()
    for fixture in fixtures:
        if not isinstance(fixture, dict):
            findings.append(finding("invalid_fixture", "each fixture must be an object"))
            continue
        fixture_id = str(fixture.get("id") or "")
        kind = str(fixture.get("kind") or "")
        if not fixture_id:
            findings.append(finding("missing_fixture_id", "fixture id is required"))
        elif fixture_id in seen_ids:
            findings.append(finding("duplicate_fixture_id", f"duplicate fixture id: {fixture_id}", fixture_id=fixture_id))
        seen_ids.add(fixture_id)
        if not kind:
            findings.append(finding("missing_fixture_kind", "fixture kind is required", fixture_id=fixture_id))
        elif kind in seen_kinds:
            findings.append(finding("duplicate_fixture_kind", f"duplicate fixture kind: {kind}", fixture_id=fixture_id))
        seen_kinds.add(kind)

        target = fixture.get("target")
        mode = target.get("mode") if isinstance(target, dict) else None
        if mode not in SUPPORTED_MODES:
            findings.append(finding("invalid_target_mode", f"unsupported target mode: {mode}", fixture_id=fixture_id))
        if mode == "extension_saved_state":
            for key in ("base_id", "extension", "state"):
                if not isinstance(target.get(key), str) or not target.get(key):
                    findings.append(finding("missing_extension_target", f"target.{key} is required", fixture_id=fixture_id))
            if required_version and target.get("platform_version") != required_version:
                findings.append(
                    finding(
                        "fixture_designer_version_mismatch",
                        f"target.platform_version must equal designer.required_version ({required_version})",
                        fixture_id=fixture_id,
                    )
                )
        if mode == "dedicated_base_active" and target.get("base_override_required") is not True:
            findings.append(finding("unsafe_shared_legacy_target", "dedicated legacy fixture must require an explicit base override", fixture_id=fixture_id))

        selector = fixture.get("selector")
        names = selector.get("names") if isinstance(selector, dict) else None
        if not isinstance(names, list) or not names or any(not isinstance(name, str) or not name.strip() for name in names):
            findings.append(finding("invalid_selector_names", "selector.names must be a non-empty string array", fixture_id=fixture_id))
        methods = fixture.get("smoke_methods")
        if not isinstance(methods, list) or not methods or any(not isinstance(method, str) or not method for method in methods):
            findings.append(finding("invalid_smoke_methods", "smoke_methods must be a non-empty string array", fixture_id=fixture_id))

    missing = sorted(set(required_kinds) - seen_kinds)
    extra = sorted(seen_kinds - set(required_kinds))
    if missing:
        findings.append(finding("required_fixture_missing", f"fixtures missing for kinds: {', '.join(missing)}"))
    if extra:
        findings.append(finding("unexpected_fixture_kind", f"unexpected fixture kinds: {', '.join(extra)}"))
    kind_paths = external_reference.get("kind_paths")
    if not isinstance(kind_paths, dict):
        findings.append(finding("missing_external_reference_kind_paths", "external_reference.kind_paths must be an object"))
    else:
        missing_reference_kinds = sorted(set(required_kinds) - set(kind_paths))
        if missing_reference_kinds:
            findings.append(
                finding(
                    "external_reference_kind_missing",
                    f"external reference paths missing for kinds: {', '.join(missing_reference_kinds)}",
                )
            )
    return findings


def normalize_guid(value: Any) -> str:
    return str(value or "").strip().strip("{}").casefold()


def matches_selector(obj: dict[str, Any], selector: dict[str, Any]) -> bool:
    names = {str(value).strip().casefold() for value in selector.get("names") or []}
    guid = normalize_guid(selector.get("guid"))
    object_name = str(obj.get("name") or "").strip().casefold()
    object_guid = normalize_guid(obj.get("guid"))
    if guid and object_guid == guid:
        return True
    return bool(object_name and object_name in names)


def object_public_selector(base_id: str, fixture: dict[str, Any], obj: dict[str, Any]) -> dict[str, Any]:
    selector: dict[str, Any] = {"base_id": base_id, "kind": fixture["kind"]}
    if obj.get("name"):
        selector["name"] = obj["name"]
    if obj.get("guid"):
        selector["guid"] = obj["guid"]
    target = fixture.get("target") or {}
    route = obj.get("route") if isinstance(obj.get("route"), dict) else {}
    extension_guid = obj.get("extension_guid") or route.get("extension_guid")
    if extension_guid:
        selector["extension_guid"] = extension_guid
    elif target.get("extension"):
        selector["extension"] = target["extension"]
    return selector


def probe_smoke_methods(
    adapter_url: str,
    service_token: str,
    timeout: float,
    fixture: dict[str, Any],
    base_id: str,
    obj: dict[str, Any],
    rpc_call: Rpc,
) -> list[dict[str, Any]]:
    selector = object_public_selector(base_id, fixture, obj)
    results: list[dict[str, Any]] = []
    for method in fixture.get("smoke_methods") or []:
        payload = {**selector, "timeout_seconds": max(1, int(timeout))}
        if method == "metadata.object.attributes":
            payload["only"] = "all"
        try:
            response = rpc_call(adapter_url, method, payload, timeout, service_token)
            status = str(response.get("status") or "unknown")
            results.append({"method": method, "status": status, **({"error": response.get("error")} if response.get("error") else {})})
        except (TimeoutError, urllib.error.URLError, OSError, ValueError, json.JSONDecodeError) as exc:
            results.append({"method": method, "status": "transport_error", "error": type(exc).__name__, "message": str(exc)[:300]})
    return results


def check_fixture_live(
    fixture: dict[str, Any],
    adapter_url: str,
    service_token: str,
    timeout: float,
    base_overrides: dict[str, str],
    rpc_call: Rpc,
    probe_cache: dict[tuple[str, str, str, str], dict[str, Any]] | None = None,
) -> dict[str, Any]:
    fixture_id = str(fixture["id"])
    target = fixture.get("target") or {}
    base_id = base_overrides.get(fixture_id) or target.get("base_id")
    if target.get("base_override_required") and fixture_id not in base_overrides:
        return {
            "id": fixture_id,
            "kind": fixture["kind"],
            "status": "target_not_configured",
            "message": f"Pass --target-base {fixture_id}=BASE_ID for the dedicated legacy fixture.",
        }
    if not base_id:
        return {"id": fixture_id, "kind": fixture["kind"], "status": "target_not_configured"}

    payload: dict[str, Any]
    method: str
    if target.get("mode") == "extension_saved_state":
        method = "extension.objects.find"
        payload = {
            "base_id": base_id,
            "extension": target["extension"],
            "state": target.get("state") or "working",
            "limit": 100,
            "use_cache": True,
            "full_scan": False,
            "refresh_cache": False,
            "timeout_seconds": max(1, int(timeout)),
        }
    else:
        method = "metadata.objects.list"
        payload = {
            "base_id": base_id,
            "kind": fixture["kind"],
            "limit": 100,
            "refresh_cache": True,
            "timeout_seconds": max(1, int(timeout)),
        }

    cache_key = (
        method,
        str(base_id),
        str(target.get("extension") or ""),
        str(target.get("state") or ""),
    )
    cached_probe_error: dict[str, Any] | None = None
    try:
        if probe_cache is not None and cache_key in probe_cache:
            cached = probe_cache[cache_key]
            cached_probe_error = cached.get("__probe_error__") if isinstance(cached.get("__probe_error__"), dict) else None
            if cached_probe_error is not None:
                raise TimeoutError(str(cached_probe_error.get("message") or "cached adapter probe failed"))
            response = cached
        else:
            response = rpc_call(adapter_url, method, payload, timeout, service_token)
            if probe_cache is not None:
                probe_cache[cache_key] = response
    except (TimeoutError, urllib.error.URLError, OSError, ValueError, json.JSONDecodeError) as exc:
        error_name = str(cached_probe_error.get("error")) if cached_probe_error else type(exc).__name__
        error_message = str(cached_probe_error.get("message")) if cached_probe_error else str(exc)[:300]
        if probe_cache is not None and cached_probe_error is None:
            probe_cache[cache_key] = {
                "__probe_error__": {
                    "error": error_name,
                    "message": error_message,
                }
            }
        return {
            "id": fixture_id,
            "kind": fixture["kind"],
            "base_id": base_id,
            "status": "adapter_error",
            "probe_method": method,
            "error": error_name,
            "message": error_message,
            **({"probe_error_reused": True} if cached_probe_error is not None else {}),
        }

    objects = response.get("objects") if isinstance(response.get("objects"), list) else response.get("items")
    objects = [obj for obj in (objects or []) if isinstance(obj, dict)]
    kind_objects = [obj for obj in objects if str(obj.get("kind") or "") == str(fixture["kind"])]
    matched = next((obj for obj in kind_objects if matches_selector(obj, fixture.get("selector") or {})), None)
    if not matched:
        return {
            "id": fixture_id,
            "kind": fixture["kind"],
            "base_id": base_id,
            "status": "fixture_missing" if response.get("status") in {"ok", "not_found"} else "adapter_error",
            "probe_method": method,
            "adapter_status": response.get("status"),
            "observed_names": sorted({str(obj.get("name")) for obj in kind_objects if obj.get("name")})[:20],
        }

    expected_states = set(fixture.get("expected_activation_states") or [])
    activation_state = str(matched.get("activation_state") or "active")
    smoke = probe_smoke_methods(adapter_url, service_token, timeout, fixture, base_id, matched, rpc_call)
    failed_smoke = [item["method"] for item in smoke if item.get("status") != "ok"]
    state_ok = not expected_states or activation_state in expected_states
    return {
        "id": fixture_id,
        "kind": fixture["kind"],
        "base_id": base_id,
        "status": "fixture_ready" if state_ok and not failed_smoke else "fixture_degraded",
        "probe_method": method,
        "adapter_status": response.get("status"),
        "object": {
            "name": matched.get("name"),
            "guid": matched.get("guid"),
            "activation_state": activation_state,
        },
        "smoke": smoke,
        **({"failed_smoke_methods": failed_smoke} if failed_smoke else {}),
        **({"unexpected_activation_state": activation_state} if not state_ok else {}),
    }


def parse_base_overrides(values: list[str]) -> dict[str, str]:
    result: dict[str, str] = {}
    for value in values:
        fixture_id, separator, base_id = value.partition("=")
        if not separator or not fixture_id.strip() or not base_id.strip():
            raise ValueError(f"invalid --target-base value: {value}; expected FIXTURE_ID=BASE_ID")
        result[fixture_id.strip()] = base_id.strip()
    return result


def check_manifest(
    manifest_path: Path,
    *,
    live: bool = False,
    adapter_url: str = "http://docker.cin.su:8011",
    service_token: str = "",
    timeout: float = 90,
    base_overrides: dict[str, str] | None = None,
    require_ready: bool = False,
    rpc_call: Rpc = rpc,
) -> dict[str, Any]:
    try:
        manifest = load_json(manifest_path)
    except (OSError, ValueError, json.JSONDecodeError) as exc:
        findings = [finding("manifest_read_failed", str(exc))]
        manifest = {}
    else:
        findings = validate_manifest(manifest)

    fixture_results: list[dict[str, Any]] = []
    if not findings:
        if live:
            probe_cache: dict[tuple[str, str, str, str], dict[str, Any]] = {}
            fixture_results = [
                check_fixture_live(
                    fixture,
                    adapter_url,
                    service_token,
                    timeout,
                    base_overrides or {},
                    rpc_call,
                    probe_cache,
                )
                for fixture in manifest.get("fixtures") or []
            ]
        else:
            fixture_results = [
                {"id": fixture["id"], "kind": fixture["kind"], "status": "not_checked"}
                for fixture in manifest.get("fixtures") or []
            ]

    ready = len([item for item in fixture_results if item.get("status") == "fixture_ready"])
    gaps = len([item for item in fixture_results if item.get("status") in {"fixture_missing", "fixture_degraded", "target_not_configured", "adapter_error"}])
    passed = not findings and (not require_ready or (bool(fixture_results) and ready == len(fixture_results)))
    return {
        "schema": REPORT_SCHEMA,
        "created_at_utc": datetime.now(timezone.utc).isoformat(),
        "status": "invalid_manifest" if findings else ("ready" if fixture_results and ready == len(fixture_results) else ("gaps_found" if live else "manifest_valid")),
        "passed": passed,
        "manifest_path": str(manifest_path),
        "live": live,
        "require_ready": require_ready,
        "adapter_url": adapter_url if live else None,
        "findings": findings,
        "fixtures": fixture_results,
        "counts": {
            "findings": len(findings),
            "fixtures": len(fixture_results),
            "ready": ready,
            "gaps": gaps,
        },
    }


def main() -> int:
    parser = argparse.ArgumentParser(description="Validate and optionally probe rare 1C metadata-kind fixtures.")
    parser.add_argument("--manifest", type=Path, default=Path("config/1c_metadata_kind_fixtures.json"))
    parser.add_argument("--live", action="store_true")
    parser.add_argument("--adapter-url", default="http://docker.cin.su:8011")
    parser.add_argument("--service-token-env", default="ONEC_ADAPTER_SERVICE_TOKEN")
    parser.add_argument("--timeout", type=float, default=90)
    parser.add_argument("--target-base", action="append", default=[], metavar="FIXTURE_ID=BASE_ID")
    parser.add_argument("--require-ready", action="store_true")
    parser.add_argument("--output", type=Path)
    args = parser.parse_args()

    try:
        base_overrides = parse_base_overrides(args.target_base)
    except ValueError as exc:
        parser.error(str(exc))
    result = check_manifest(
        args.manifest,
        live=args.live,
        adapter_url=args.adapter_url,
        service_token=os.environ.get(args.service_token_env, ""),
        timeout=args.timeout,
        base_overrides=base_overrides,
        require_ready=args.require_ready,
    )
    if args.output:
        write_json(args.output, result)
    print(json.dumps(result, ensure_ascii=False, indent=2))
    return 0 if result["passed"] else 2


if __name__ == "__main__":
    raise SystemExit(main())