385 lines
16 KiB
Python
385 lines
16 KiB
Python
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"<li><strong>{esc(item.get('severity'))}: {esc(item.get('code'))}</strong><br>{esc(item.get('message'))}</li>"
|
||
for item in findings
|
||
if isinstance(item, dict)
|
||
) or "<li>Findings absent.</li>"
|
||
table_rows = "\n".join(
|
||
"<tr>"
|
||
f"<td>{esc(row.get('matched_role_name'))}</td>"
|
||
f"<td>{esc(row.get('profile_name'))}</td>"
|
||
f"<td>{esc(row.get('group_name'))}</td>"
|
||
f"<td>{esc(row.get('user_name'))}</td>"
|
||
f"<td>{esc(row.get('user_type'))}</td>"
|
||
f"<td>{esc(row.get('user_active'))}</td>"
|
||
f"<td>{esc(row.get('access_path'))}</td>"
|
||
"</tr>"
|
||
for row in rows
|
||
if isinstance(row, dict)
|
||
)
|
||
artifact_links = "\n".join(
|
||
f"<a href=\"{esc(Path(path).name)}\">{esc(name)}</a>"
|
||
for name, path in artifacts.items()
|
||
)
|
||
return f"""<!doctype html>
|
||
<html lang="ru">
|
||
<head>
|
||
<meta charset="utf-8">
|
||
<title>1C Access Role Audit</title>
|
||
<style>
|
||
body {{ font-family: Segoe UI, Arial, sans-serif; margin: 24px; color: #1f2933; }}
|
||
h1 {{ margin-bottom: 4px; }}
|
||
.meta, .artifacts {{ color: #52616b; margin: 8px 0 18px; }}
|
||
.badge {{ display: inline-block; padding: 4px 10px; border-radius: 999px; font-weight: 700; }}
|
||
.low {{ background: #e3f8e5; color: #176b32; }}
|
||
.medium {{ background: #fff2cc; color: #875a00; }}
|
||
.high {{ background: #ffe0e0; color: #9f1d1d; }}
|
||
.cards {{ display: flex; flex-wrap: wrap; gap: 12px; margin: 18px 0; }}
|
||
.card {{ border: 1px solid #d9e2ec; border-radius: 8px; padding: 12px 14px; min-width: 120px; }}
|
||
.card b {{ display: block; font-size: 20px; }}
|
||
input {{ width: 100%; max-width: 520px; padding: 9px 10px; margin: 12px 0; border: 1px solid #bcccdc; border-radius: 6px; }}
|
||
table {{ border-collapse: collapse; width: 100%; margin-top: 12px; }}
|
||
th, td {{ border: 1px solid #d9e2ec; padding: 8px; vertical-align: top; }}
|
||
th {{ background: #f0f4f8; text-align: left; position: sticky; top: 0; }}
|
||
tr:nth-child(even) {{ background: #fbfcfd; }}
|
||
.artifacts a {{ margin-right: 12px; }}
|
||
</style>
|
||
</head>
|
||
<body>
|
||
<h1>1C Access Role Audit</h1>
|
||
<div class="meta">Base: <b>{esc(summary.get('base_id'))}</b> | Role query: <b>{esc(role)}</b> | Generated: {esc(summary.get('generated_at'))}</div>
|
||
<div>Risk: <span class="badge {esc(risk)}">{esc(risk)}</span></div>
|
||
<div class="cards">
|
||
<div class="card"><b>{esc(counts.get('users'))}</b>users</div>
|
||
<div class="card"><b>{esc(counts.get('groups'))}</b>groups</div>
|
||
<div class="card"><b>{esc(counts.get('profiles'))}</b>profiles</div>
|
||
<div class="card"><b>{esc(counts.get('rows'))}</b>rows</div>
|
||
<div class="card"><b>{esc(len(findings))}</b>findings</div>
|
||
</div>
|
||
<h2>Findings</h2>
|
||
<ul>{finding_items}</ul>
|
||
<div class="artifacts">Artifacts: {artifact_links}</div>
|
||
<h2>Rows</h2>
|
||
<input id="filter" placeholder="Filter by user, group, profile, role..." oninput="filterRows()">
|
||
<table id="rows">
|
||
<thead><tr><th>Role</th><th>Profile</th><th>Group</th><th>User</th><th>Type</th><th>Active</th><th>Access path</th></tr></thead>
|
||
<tbody>{table_rows}</tbody>
|
||
</table>
|
||
<script>
|
||
function filterRows() {{
|
||
const q = document.getElementById('filter').value.toLowerCase();
|
||
document.querySelectorAll('#rows tbody tr').forEach(tr => {{
|
||
tr.style.display = tr.innerText.toLowerCase().includes(q) ? '' : 'none';
|
||
}});
|
||
}}
|
||
</script>
|
||
</body>
|
||
</html>
|
||
"""
|
||
|
||
|
||
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"<a href=\"{esc(compare_name)}\">compare_previous</a>" if compare_name and (Path(str(item.get("summary_path"))).parent / compare_name).exists() else ""
|
||
links = " ".join(
|
||
f"<a href=\"{esc(Path(path).name)}\">{esc(name)}</a>"
|
||
for name, path in artifacts.items()
|
||
if name != "index"
|
||
) + (f" {compare_link}" if compare_link else "")
|
||
rows.append(
|
||
"<tr>"
|
||
f"<td>{esc(item.get('generated_at'))}</td>"
|
||
f"<td>{esc(query.get('role') if isinstance(query, dict) else None)}</td>"
|
||
f"<td><span class=\"badge {esc(item.get('risk_level') or 'unknown')}\">{esc(item.get('risk_level') or 'unknown')}</span></td>"
|
||
f"<td>{esc(counts.get('users'))}</td>"
|
||
f"<td>{esc(counts.get('rows'))}</td>"
|
||
f"<td>{esc(len(findings))}</td>"
|
||
f"<td>{links}</td>"
|
||
"</tr>"
|
||
)
|
||
table_rows = "\n".join(rows) or "<tr><td colspan=\"7\">Reports absent.</td></tr>"
|
||
generated_at = datetime.now(timezone.utc).isoformat()
|
||
return f"""<!doctype html>
|
||
<html lang="ru">
|
||
<head>
|
||
<meta charset="utf-8">
|
||
<title>1C Access Audit Index</title>
|
||
<style>
|
||
body {{ font-family: Segoe UI, Arial, sans-serif; margin: 24px; color: #1f2933; }}
|
||
h1 {{ margin-bottom: 4px; }}
|
||
.meta {{ color: #52616b; margin: 8px 0 18px; }}
|
||
input {{ width: 100%; max-width: 560px; padding: 9px 10px; margin: 12px 0; border: 1px solid #bcccdc; border-radius: 6px; }}
|
||
table {{ border-collapse: collapse; width: 100%; margin-top: 12px; }}
|
||
th, td {{ border: 1px solid #d9e2ec; padding: 8px; vertical-align: top; }}
|
||
th {{ background: #f0f4f8; text-align: left; position: sticky; top: 0; }}
|
||
tr:nth-child(even) {{ background: #fbfcfd; }}
|
||
a {{ margin-right: 10px; }}
|
||
.badge {{ display: inline-block; padding: 3px 9px; border-radius: 999px; font-weight: 700; }}
|
||
.low {{ background: #e3f8e5; color: #176b32; }}
|
||
.medium {{ background: #fff2cc; color: #875a00; }}
|
||
.high {{ background: #ffe0e0; color: #9f1d1d; }}
|
||
.unknown {{ background: #edf2f7; color: #52616b; }}
|
||
</style>
|
||
</head>
|
||
<body>
|
||
<h1>1C Access Audit Index</h1>
|
||
<div class="meta">Base: <b>{esc(base_id)}</b> | Reports: <b>{esc(len(summaries))}</b> | Updated: {esc(generated_at)}</div>
|
||
<input id="filter" placeholder="Filter by role, risk, artifact..." oninput="filterRows()">
|
||
<table id="reports">
|
||
<thead><tr><th>Generated</th><th>Role query</th><th>Risk</th><th>Users</th><th>Rows</th><th>Findings</th><th>Artifacts</th></tr></thead>
|
||
<tbody>{table_rows}</tbody>
|
||
</table>
|
||
<script>
|
||
function filterRows() {{
|
||
const q = document.getElementById('filter').value.toLowerCase();
|
||
document.querySelectorAll('#reports tbody tr').forEach(tr => {{
|
||
tr.style.display = tr.innerText.toLowerCase().includes(q) ? '' : 'none';
|
||
}});
|
||
}}
|
||
</script>
|
||
</body>
|
||
</html>
|
||
"""
|
||
|
||
|
||
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())
|