llm/scripts/audit_1c_access_role.py

from __future__ import annotations

import argparse
import json
import sys
from pathlib import Path
from typing import Any

ROOT = Path(__file__).resolve().parents[1]
if str(ROOT) not in sys.path:
    sys.path.insert(0, str(ROOT))

import scripts.compare_1c_access_role_audit as compare_script
import scripts.export_1c_access_role_audit as export_script


def compare_stem(old_path: Path, new_path: Path) -> str:
    return f"compare-{old_path.stem.replace('.summary', '')}-{new_path.stem.replace('.summary', '')}"


def verdict_for(export_summary: dict[str, Any], compare_result: dict[str, Any] | None) -> str:
    counts = compare_result.get("counts") if isinstance(compare_result, dict) and isinstance(compare_result.get("counts"), dict) else {}
    if any(int(counts.get(key) or 0) > 0 for key in ("added_users", "removed_users", "changed_access_paths")):
        return "changed"
    if str(export_summary.get("risk_level") or "").lower() in {"medium", "high"}:
        return "risk"
    return "ok"


def audit_role(
    *,
    adapter_url: str,
    base_id: str,
    role: str,
    report_root: Path,
    timeout: int,
    user_threshold: int,
    limit: int,
    include_html: bool = True,
) -> dict[str, Any]:
    export_summary = export_script.export_role_audit(
        adapter_url=adapter_url,
        base_id=base_id,
        role=role,
        report_root=report_root,
        timeout=timeout,
        user_threshold=user_threshold,
        include_analysis=True,
        limit=limit,
        include_html=include_html,
    )
    latest = compare_script.find_latest_summaries(report_root, base_id, role=role, count=2)
    compare_result: dict[str, Any] | None = None
    if len(latest) >= 2:
        new_path, old_path = latest[0], latest[1]
        folder = report_root / compare_script.slugify(base_id, max_length=60)
        stem = compare_stem(old_path, new_path)
        compare_result = compare_script.compare_files(
            old_path,
            new_path,
            output=folder / f"{stem}.json",
            html_output=folder / f"{stem}.html",
        )
        export_script.update_index(report_root, base_id)
    return {
        "schema": "onec_access_role_audit_run.v1",
        "status": export_summary.get("status"),
        "base_id": base_id,
        "role": role,
        "verdict": verdict_for(export_summary, compare_result),
        "export": export_summary,
        "compare": compare_result,
    }


def audit_config(
    *,
    config_path: Path,
    adapter_url: str,
    report_root: Path,
    timeout: int,
    limit: int,
    include_html: bool = True,
) -> dict[str, Any]:
    config = json.loads(config_path.read_text(encoding="utf-8"))
    roles = config.get("roles") if isinstance(config.get("roles"), list) else []
    default_base_id = str(config.get("base_id") or "upo_test")
    results: list[dict[str, Any]] = []
    for item in roles:
        if not isinstance(item, dict) or not item.get("role"):
            continue
        results.append(
            audit_role(
                adapter_url=adapter_url,
                base_id=str(item.get("base_id") or default_base_id),
                role=str(item.get("role")),
                report_root=report_root,
                timeout=timeout,
                user_threshold=int(item.get("user_threshold") or 50),
                limit=limit,
                include_html=include_html,
            )
        )
    verdicts = [str(item.get("verdict") or "") for item in results]
    overall = "changed" if "changed" in verdicts else "risk" if "risk" in verdicts else "ok"
    return {
        "schema": "onec_access_role_audit_batch.v1",
        "status": "ok" if all(item.get("status") == "ok" for item in results) else "error",
        "config": str(config_path),
        "overall_verdict": overall,
        "counts": {"roles": len(results), "changed": verdicts.count("changed"), "risk": verdicts.count("risk"), "ok": verdicts.count("ok")},
        "results": results,
    }


def main() -> int:
    parser = argparse.ArgumentParser(description="Run 1C role access audit: export, analyze, update index, compare with previous.")
    parser.add_argument("--adapter-url", default=export_script.DEFAULT_BASE_URL)
    parser.add_argument("--base-id", default="upo_test")
    parser.add_argument("--role")
    parser.add_argument("--config", type=Path, help="Run all roles from an access critical roles JSON config.")
    parser.add_argument("--report-root", type=Path, default=export_script.DEFAULT_REPORT_ROOT)
    parser.add_argument("--timeout", type=int, default=120)
    parser.add_argument("--limit", type=int, default=20000)
    parser.add_argument("--user-threshold", type=int, default=50)
    parser.add_argument("--no-html", action="store_true")
    args = parser.parse_args()

    if args.config:
        result = audit_config(
            config_path=args.config,
            adapter_url=args.adapter_url,
            report_root=args.report_root,
            timeout=args.timeout,
            limit=args.limit,
            include_html=not args.no_html,
        )
    else:
        if not args.role:
            parser.error("--role is required unless --config is used.")
        result = audit_role(
            adapter_url=args.adapter_url,
            base_id=args.base_id,
            role=args.role,
            report_root=args.report_root,
            timeout=args.timeout,
            user_threshold=args.user_threshold,
            limit=args.limit,
            include_html=not args.no_html,
        )
    print(json.dumps(result, ensure_ascii=False, indent=2))
    return 0 if result.get("status") == "ok" else 1


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