195 lines
9.0 KiB
Python
195 lines
9.0 KiB
Python
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())
|