from __future__ import annotations import argparse import hashlib import html import json import re import sys import urllib.request from datetime import datetime, timezone from pathlib import Path from typing import Any ROOT = Path(__file__).resolve().parents[1] DEFAULT_BASE_URL = "http://docker.cin.su:8011" DEFAULT_REPORT_ROOT = ROOT / "reports" / "1c-access" def slugify(value: str, *, max_length: int = 80) -> str: slug = re.sub(r"[^0-9A-Za-zА-Яа-яЁё._-]+", "-", value.strip()) slug = re.sub(r"-+", "-", slug).strip("-._") return (slug or "role-audit")[:max_length] def rpc(base_url: str, method: str, payload: dict[str, Any], *, timeout: int) -> dict[str, Any]: body = json.dumps({"method": method, "payload": payload}, ensure_ascii=False).encode("utf-8") request = urllib.request.Request( base_url.rstrip("/") + "/rpc", data=body, headers={"Content-Type": "application/json"}, method="POST", ) with urllib.request.urlopen(request, timeout=timeout) as response: data = json.loads(response.read().decode("utf-8")) if not isinstance(data, dict): return {"status": "error", "error": "response_not_object", "response": data} return data 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), encoding="utf-8") def build_artifact_paths(report_root: Path, base_id: str, role: str, timestamp: str) -> dict[str, Path]: digest = hashlib.sha1(role.encode("utf-8")).hexdigest()[:10] stem = f"{timestamp}-{slugify(role)}-{digest}" folder = report_root / slugify(base_id, max_length=60) return { "csv": folder / f"{stem}.csv", "json": folder / f"{stem}.json", "analysis": folder / f"{stem}.analysis.json", "summary": folder / f"{stem}.summary.json", "html": folder / f"{stem}.html", "index": folder / "index.html", } def esc(value: Any) -> str: return html.escape("" if value is None else str(value), quote=True) def render_html_report(summary: dict[str, Any], export: dict[str, Any], analysis: dict[str, Any] | None) -> str: rows = export.get("rows") if isinstance(export.get("rows"), list) else [] findings = summary.get("findings") if isinstance(summary.get("findings"), list) else [] counts = summary.get("counts") if isinstance(summary.get("counts"), dict) else {} artifacts = summary.get("artifacts") if isinstance(summary.get("artifacts"), dict) else {} risk = str(summary.get("risk_level") or "unknown") role = ((summary.get("query") if isinstance(summary.get("query"), dict) else {}) or {}).get("role") finding_items = "\n".join( f"
  • {esc(item.get('severity'))}: {esc(item.get('code'))}
    {esc(item.get('message'))}
  • " for item in findings if isinstance(item, dict) ) or "
  • Findings absent.
  • " table_rows = "\n".join( "" f"{esc(row.get('matched_role_name'))}" f"{esc(row.get('profile_name'))}" f"{esc(row.get('group_name'))}" f"{esc(row.get('user_name'))}" f"{esc(row.get('user_type'))}" f"{esc(row.get('user_active'))}" f"{esc(row.get('access_path'))}" "" for row in rows if isinstance(row, dict) ) artifact_links = "\n".join( f"{esc(name)}" for name, path in artifacts.items() ) return f""" 1C Access Role Audit

    1C Access Role Audit

    Base: {esc(summary.get('base_id'))} | Role query: {esc(role)} | Generated: {esc(summary.get('generated_at'))}
    Risk: {esc(risk)}
    {esc(counts.get('users'))}users
    {esc(counts.get('groups'))}groups
    {esc(counts.get('profiles'))}profiles
    {esc(counts.get('rows'))}rows
    {esc(len(findings))}findings

    Findings

    Artifacts: {artifact_links}

    Rows

    {table_rows}
    RoleProfileGroupUserTypeActiveAccess path
    """ def load_audit_summaries(folder: Path, *, limit: int = 100) -> list[dict[str, Any]]: summaries: list[dict[str, Any]] = [] for path in folder.glob("*.summary.json"): try: data = json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): continue if isinstance(data, dict): data.setdefault("summary_path", str(path)) summaries.append(data) summaries.sort(key=lambda item: str(item.get("generated_at") or ""), reverse=True) return summaries[:limit] def summary_role(summary: dict[str, Any]) -> str | None: query = summary.get("query") if isinstance(summary.get("query"), dict) else {} role = query.get("role") if isinstance(query, dict) else None return str(role) if role is not None else None def compare_filename_for_pair(old_summary: dict[str, Any], new_summary: dict[str, Any], *, suffix: str) -> str | None: old_path = old_summary.get("summary_path") new_path = new_summary.get("summary_path") if not old_path or not new_path: return None old_stem = Path(str(old_path)).stem.replace(".summary", "") new_stem = Path(str(new_path)).stem.replace(".summary", "") return f"compare-{old_stem}-{new_stem}.{suffix}" def render_index_html(base_id: str, summaries: list[dict[str, Any]]) -> str: rows: list[str] = [] previous_by_role: dict[str, dict[str, Any]] = {} for item in reversed(summaries): role_key = (summary_role(item) or "").casefold() item["previous_summary_path"] = previous_by_role.get(role_key, {}).get("summary_path") previous_by_role[role_key] = item for item in summaries: query = item.get("query") if isinstance(item.get("query"), dict) else {} counts = item.get("counts") if isinstance(item.get("counts"), dict) else {} findings = item.get("findings") if isinstance(item.get("findings"), list) else [] artifacts = item.get("artifacts") if isinstance(item.get("artifacts"), dict) else {} previous = next((candidate for candidate in summaries if candidate.get("summary_path") == item.get("previous_summary_path")), None) compare_name = compare_filename_for_pair(previous, item, suffix="html") if isinstance(previous, dict) else None compare_link = f"compare_previous" if compare_name and (Path(str(item.get("summary_path"))).parent / compare_name).exists() else "" links = " ".join( f"{esc(name)}" for name, path in artifacts.items() if name != "index" ) + (f" {compare_link}" if compare_link else "") rows.append( "" f"{esc(item.get('generated_at'))}" f"{esc(query.get('role') if isinstance(query, dict) else None)}" f"{esc(item.get('risk_level') or 'unknown')}" f"{esc(counts.get('users'))}" f"{esc(counts.get('rows'))}" f"{esc(len(findings))}" f"{links}" "" ) table_rows = "\n".join(rows) or "Reports absent." generated_at = datetime.now(timezone.utc).isoformat() return f""" 1C Access Audit Index

    1C Access Audit Index

    Base: {esc(base_id)} | Reports: {esc(len(summaries))} | Updated: {esc(generated_at)}
    {table_rows}
    GeneratedRole queryRiskUsersRowsFindingsArtifacts
    """ def update_index(report_root: Path, base_id: str, *, limit: int = 100) -> Path: folder = report_root / slugify(base_id, max_length=60) folder.mkdir(parents=True, exist_ok=True) index_path = folder / "index.html" index_path.write_text(render_index_html(base_id, load_audit_summaries(folder, limit=limit)), encoding="utf-8") return index_path def export_role_audit( *, adapter_url: str, base_id: str, role: str, report_root: Path, timeout: int, user_threshold: int, include_analysis: bool, limit: int, include_html: bool = True, ) -> dict[str, Any]: export = rpc( adapter_url, "access.role.audit_export", { "base_id": base_id, "role": role, "format": "csv", "limit": limit, "timeout_seconds": timeout, }, timeout=timeout + 30, ) analysis: dict[str, Any] | None = None if include_analysis: analysis = rpc( adapter_url, "access.role.audit_analyze", { "base_id": base_id, "role": role, "user_threshold": user_threshold, "timeout_seconds": timeout, }, timeout=timeout + 30, ) timestamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") paths = build_artifact_paths(report_root, base_id, role, timestamp) paths["csv"].parent.mkdir(parents=True, exist_ok=True) paths["csv"].write_text(str(export.get("csv") or ""), encoding="utf-8-sig") write_json(paths["json"], export) if analysis is not None: write_json(paths["analysis"], analysis) summary = { "schema": "onec_access_role_audit_artifact.v1", "status": "ok" if export.get("status") == "ok" and (analysis is None or analysis.get("status") == "ok") else "error", "base_id": base_id, "query": {"role": role}, "generated_at": datetime.now(timezone.utc).isoformat(), "adapter_url": adapter_url, "artifacts": {key: str(path) for key, path in paths.items() if key != "analysis" or analysis is not None}, "counts": export.get("counts") if isinstance(export.get("counts"), dict) else {}, "risk_level": analysis.get("risk_level") if isinstance(analysis, dict) else None, "findings": analysis.get("findings") if isinstance(analysis, dict) else [], "export_status": export.get("status"), "analysis_status": analysis.get("status") if isinstance(analysis, dict) else None, } if include_html: paths["html"].write_text(render_html_report(summary, export, analysis), encoding="utf-8") else: summary["artifacts"].pop("html", None) write_json(paths["summary"], summary) update_index(report_root, base_id) return summary def main() -> int: parser = argparse.ArgumentParser(description="Export 1C/BSP role access audit artifacts to the local workspace.") parser.add_argument("--adapter-url", default=DEFAULT_BASE_URL) parser.add_argument("--base-id", default="upo_test") parser.add_argument("--role", required=True) parser.add_argument("--report-root", type=Path, default=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-analysis", action="store_true") parser.add_argument("--no-html", action="store_true") parser.add_argument("--json", action="store_true", help="Print full artifact summary JSON.") args = parser.parse_args() summary = export_role_audit( 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, include_analysis=not args.no_analysis, limit=args.limit, include_html=not args.no_html, ) if args.json: print(json.dumps(summary, ensure_ascii=False, indent=2)) else: print( json.dumps( { "status": summary.get("status"), "base_id": summary.get("base_id"), "role": role_summary(summary), "risk_level": summary.get("risk_level"), "counts": summary.get("counts"), "artifacts": summary.get("artifacts"), }, ensure_ascii=False, indent=2, ) ) return 0 if summary.get("status") == "ok" else 1 def role_summary(summary: dict[str, Any]) -> str | None: query = summary.get("query") if isinstance(summary.get("query"), dict) else {} return query.get("role") if isinstance(query, dict) else None if __name__ == "__main__": raise SystemExit(main())