from __future__ import annotations import argparse import json import sys import time import urllib.request from pathlib import Path from typing import Any ROOT = Path(__file__).resolve().parents[1] DEFAULT_BASE_URL = "http://docker.cin.su:8011" DEFAULT_REF = "РегистрСведений.УОП_АктуальныеСпецификации" DEFAULT_REPORT = ROOT / "reports" / "1c-access" / "upo_test-access-object.json" 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: result = json.loads(response.read().decode("utf-8")) if not isinstance(result, dict): return {"schema": "onec_access_object_smoke_error.v1", "status": "error", "error": "response_not_object", "response": result} return result def compact_roles(result: dict[str, Any], limit: int = 10) -> list[str]: return [ str(item.get("name") or item.get("id") or "") for item in result.get("roles") or [] if isinstance(item, dict) ][:limit] def compact_names(result: dict[str, Any], key: str, limit: int = 10) -> list[str]: return [ str(item.get("name") or item.get("id") or item.get(key) or "") for item in result.get(key) or [] if isinstance(item, dict) ][:limit] def main() -> int: parser = argparse.ArgumentParser(description="Smoke-check 1C access roles/subjects by metadata object name/ref.") parser.add_argument("--adapter-url", default=DEFAULT_BASE_URL) parser.add_argument("--base-id", default="upo_test") parser.add_argument("--ref", default=DEFAULT_REF) parser.add_argument("--action", default="write") parser.add_argument("--limit", type=int, default=10) parser.add_argument("--max-effective-per-user", type=int, default=0) parser.add_argument("--timeout", type=int, default=420) parser.add_argument("--include-subjects", action="store_true", help="Also call access.object.subjects; this can be heavy on large bases.") parser.add_argument("--include-access-key-scope", action="store_true", help="When --include-subjects is set, also request BSP subject access key scope.") parser.add_argument("--access-key-scope-limit", type=int, default=20000, help="Maximum rows per BSP access-key extractor when --include-access-key-scope is set.") parser.add_argument("--access-key-scope-subject-limit", type=int, default=20000, help="Maximum matched groups/users included in scope diagnostics; independent from --limit.") parser.add_argument("--include-rls-discover", action="store_true", help="Also verify access.rls.discover returns candidates.") parser.add_argument("--report", type=Path, default=DEFAULT_REPORT) parser.add_argument("--json", action="store_true", help="Print full smoke report.") args = parser.parse_args() report: dict[str, Any] = { "schema": "onec_access_object_smoke.v1", "base_id": args.base_id, "ref": args.ref, "action": args.action, "passed": False, "steps": [], } required_methods = {"access.object.roles", "access.object.subjects", "access.rls.discover"} try: help_result = rpc(args.adapter_url, "help.methods", {}, timeout=min(args.timeout, 60)) methods_text = json.dumps(help_result, ensure_ascii=False) visible = {method: method in methods_text for method in sorted(required_methods)} report["steps"].append({"name": "help.methods", "status": "ok", "visible": visible}) if not visible["access.object.roles"]: report["issues"] = [{"code": "access_object_roles_missing"}] return write_and_exit(report, args.report, args.json) roles_payload = { "base_id": args.base_id, "ref": args.ref, "action": args.action, "limit": args.limit, "timeout_seconds": min(args.timeout, 120), "max_effective_permissions_per_user": args.max_effective_per_user, } started = time.monotonic() roles = rpc(args.adapter_url, "access.object.roles", roles_payload, timeout=args.timeout) report["steps"].append( { "name": "access.object.roles", "duration_seconds": round(time.monotonic() - started, 1), "schema": roles.get("schema"), "status": roles.get("status"), "counts": roles.get("counts"), "object": roles.get("object"), "roles": compact_roles(roles, args.limit), } ) if roles.get("status") != "ok" or int(((roles.get("counts") or {}).get("roles") or 0)) <= 0: report["issues"] = [{"code": "roles_not_found", "status": roles.get("status"), "counts": roles.get("counts")}] return write_and_exit(report, args.report, args.json) if args.include_subjects: subjects_payload = dict(roles_payload) if args.include_access_key_scope: subjects_payload["include_access_key_scope"] = True subjects_payload["access_key_scope_limit"] = args.access_key_scope_limit subjects_payload["access_key_scope_subject_limit"] = args.access_key_scope_subject_limit started = time.monotonic() subjects = rpc(args.adapter_url, "access.object.subjects", subjects_payload, timeout=args.timeout) access_key_scope = subjects.get("access_key_scope") if isinstance(subjects.get("access_key_scope"), dict) else None report["steps"].append( { "name": "access.object.subjects", "duration_seconds": round(time.monotonic() - started, 1), "schema": subjects.get("schema"), "status": subjects.get("status"), "counts": subjects.get("counts"), "profiles": compact_names(subjects, "profiles", args.limit), "groups": compact_names(subjects, "groups", args.limit), "users": compact_names(subjects, "users", args.limit), **( { "access_key_scope": { "status": access_key_scope.get("status"), "counts": access_key_scope.get("counts"), "coverage": access_key_scope.get("coverage"), "truncated": access_key_scope.get("truncated"), } } if access_key_scope else {} ), } ) if args.include_rls_discover: started = time.monotonic() rls = rpc(args.adapter_url, "access.rls.discover", {"base_id": args.base_id, "terms": ["Доступ"], "limit": 3, "timeout_seconds": min(args.timeout, 120)}, timeout=args.timeout) report["steps"].append( { "name": "access.rls.discover", "duration_seconds": round(time.monotonic() - started, 1), "schema": rls.get("schema"), "status": rls.get("status"), "counts": rls.get("counts"), "candidates": [ ".".join(str(part) for part in (item.get("kind"), item.get("name")) if part) for item in rls.get("candidates") or [] if isinstance(item, dict) ], } ) report["passed"] = True return write_and_exit(report, args.report, args.json) except Exception as exc: report["issues"] = [{"code": "exception", "type": type(exc).__name__, "message": str(exc)}] return write_and_exit(report, args.report, args.json) def write_and_exit(report: dict[str, Any], report_path: Path, print_json: bool) -> int: report_path.parent.mkdir(parents=True, exist_ok=True) report_path.write_text(json.dumps(report, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") if print_json: print(json.dumps(report, ensure_ascii=False, indent=2)) else: last_step = report["steps"][-1] if report.get("steps") else {} print( json.dumps( { "passed": report.get("passed"), "last_step": last_step.get("name"), "last_status": last_step.get("status"), "counts": last_step.get("counts"), "issues": report.get("issues") or [], "report": str(report_path), }, ensure_ascii=False, indent=2, ) ) return 0 if report.get("passed") else 1 if __name__ == "__main__": raise SystemExit(main())