#!/usr/bin/env python3 from __future__ import annotations import argparse import csv import gzip import json import shutil import subprocess import tempfile from collections import Counter from pathlib import Path from typing import Any try: import pymysql except ImportError: # pragma: no cover - optional dependency for execution only pymysql = None TARGET_COLUMNS = [ "PermissionsDelete", "PermissionsViewAllRecords", "PermissionsModifyAllRecords", ] DEFAULT_SAMPLE_LIMIT = 20 DEFAULT_FOCUS_PARENT_IDS = [ "0PS5g000001HHVSGA4", ] def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser( description="Backfill ObjectPermissions D/VA/MA columns for profile-owned shadow permission sets from a frozen Salesforce snapshot." ) source = parser.add_mutually_exclusive_group(required=True) source.add_argument("--snapshot-dir", help="Local snapshot directory containing ObjectPermissions.csv.gz.") source.add_argument("--s3-uri", help="Direct S3 URI to ObjectPermissions.csv.gz.") parser.add_argument("--db-host", required=True) parser.add_argument("--db-port", type=int, default=4000) parser.add_argument("--db-user", required=True) parser.add_argument("--db-password", required=True) parser.add_argument("--db-database", required=True) parser.add_argument("--apply", action="store_true", help="Execute UPDATEs. Default is dry-run.") parser.add_argument("--sample-limit", type=int, default=DEFAULT_SAMPLE_LIMIT, help="Max diff samples to print.") parser.add_argument( "--focus-parent-id", action="append", dest="focus_parent_ids", default=[], help="ParentId to highlight in the report. May be passed multiple times.", ) return parser.parse_args() def require_aws_cli() -> str: aws_path = shutil.which("aws") if not aws_path: raise RuntimeError("aws CLI is required for --s3-uri but was not found in PATH") return aws_path def download_s3_file(s3_uri: str) -> Path: aws_path = require_aws_cli() temp_dir = Path(tempfile.mkdtemp(prefix="sf-objectpermissions-backfill-")) target_path = temp_dir / "ObjectPermissions.csv.gz" subprocess.run([aws_path, "s3", "cp", s3_uri, str(target_path)], check=True) return target_path def normalize_bool(raw: str) -> int: value = str(raw or "").strip().lower() return 1 if value in {"1", "true", "t", "yes", "y"} else 0 def load_source_rows(csv_path: Path) -> dict[tuple[str, str], dict[str, Any]]: rows: dict[tuple[str, str], dict[str, Any]] = {} with gzip.open(csv_path, "rt", encoding="utf-8-sig", newline="") as src: for row in csv.DictReader(src): parent_id = (row.get("ParentId") or "").strip() sobject_type = (row.get("SobjectType") or "").strip() if not parent_id or not sobject_type: continue rows[(parent_id, sobject_type)] = { "ParentId": parent_id, "SobjectType": sobject_type, "PermissionsDelete": normalize_bool(row.get("PermissionsDelete") or ""), "PermissionsViewAllRecords": normalize_bool(row.get("PermissionsViewAllRecords") or ""), "PermissionsModifyAllRecords": normalize_bool(row.get("PermissionsModifyAllRecords") or ""), } return rows def source_csv_path(args: argparse.Namespace) -> Path: if args.snapshot_dir: return Path(args.snapshot_dir).expanduser().resolve() / "ObjectPermissions.csv.gz" return download_s3_file(str(args.s3_uri)) def fetch_target_rows(conn: pymysql.connections.Connection) -> list[dict[str, Any]]: sql = """ SELECT op.Id, op.ParentId, op.SobjectType, IFNULL(op.PermissionsDelete, 0), IFNULL(op.PermissionsViewAllRecords, 0), IFNULL(op.PermissionsModifyAllRecords, 0) FROM `ObjectPermissions` op JOIN `PermissionSet` ps ON ps.Id = op.ParentId WHERE IFNULL(ps.IsOwnedByProfile, 0) = 1 """ with conn.cursor() as cur: cur.execute(sql) return [ { "Id": row[0], "ParentId": row[1], "SobjectType": row[2], "PermissionsDelete": int(row[3]), "PermissionsViewAllRecords": int(row[4]), "PermissionsModifyAllRecords": int(row[5]), } for row in cur.fetchall() ] def build_plan(target_rows: list[dict[str, Any]], source_rows: dict[tuple[str, str], dict[str, Any]]) -> tuple[list[dict[str, Any]], dict[str, Any]]: updates: list[dict[str, Any]] = [] missing_source = 0 unchanged = 0 per_parent = Counter() for target in target_rows: key = (str(target["ParentId"]), str(target["SobjectType"])) source = source_rows.get(key) if source is None: missing_source += 1 continue before = {name: int(target[name]) for name in TARGET_COLUMNS} after = {name: int(source[name]) for name in TARGET_COLUMNS} if before == after: unchanged += 1 continue diff = { "Id": target["Id"], "ParentId": target["ParentId"], "SobjectType": target["SobjectType"], "before": before, "after": after, } updates.append(diff) per_parent[str(target["ParentId"])] += 1 summary = { "target_shadow_rows": len(target_rows), "source_rows": len(source_rows), "planned_updates": len(updates), "unchanged_rows": unchanged, "missing_source_rows": missing_source, "per_parent_updates": dict(sorted(per_parent.items())), } return updates, summary def build_focus_summary(updates: list[dict[str, Any]], focus_parent_ids: list[str]) -> dict[str, list[dict[str, Any]]]: if not focus_parent_ids: return {} wanted = {item.strip() for item in focus_parent_ids if item.strip()} if not wanted: return {} result: dict[str, list[dict[str, Any]]] = {} for item in updates: parent_id = str(item["ParentId"]) if parent_id not in wanted: continue result.setdefault(parent_id, []).append( { "SobjectType": item["SobjectType"], "before": item["before"], "after": item["after"], } ) return result def apply_updates(conn: pymysql.connections.Connection, updates: list[dict[str, Any]]) -> int: if not updates: return 0 sql = """ UPDATE `ObjectPermissions` SET `PermissionsDelete` = %s, `PermissionsViewAllRecords` = %s, `PermissionsModifyAllRecords` = %s WHERE `Id` = %s AND ( IFNULL(`PermissionsDelete`, 0) <> %s OR IFNULL(`PermissionsViewAllRecords`, 0) <> %s OR IFNULL(`PermissionsModifyAllRecords`, 0) <> %s ) """ with conn.cursor() as cur: cur.executemany( sql, [ ( item["after"]["PermissionsDelete"], item["after"]["PermissionsViewAllRecords"], item["after"]["PermissionsModifyAllRecords"], item["Id"], item["after"]["PermissionsDelete"], item["after"]["PermissionsViewAllRecords"], item["after"]["PermissionsModifyAllRecords"], ) for item in updates ], ) return cur.rowcount def main() -> int: args = parse_args() if pymysql is None: raise RuntimeError("PyMySQL is required. Install it before running this backfill script.") focus_parent_ids = list(dict.fromkeys(DEFAULT_FOCUS_PARENT_IDS + list(args.focus_parent_ids or []))) csv_path = source_csv_path(args) if not csv_path.exists(): raise FileNotFoundError(f"ObjectPermissions source file not found: {csv_path}") source_rows = load_source_rows(csv_path) conn = pymysql.connect( host=args.db_host, port=args.db_port, user=args.db_user, password=args.db_password, database=args.db_database, ssl={"ssl": {}}, charset="utf8mb4", autocommit=False, ) report: dict[str, Any] = { "dry_run": not args.apply, "source": str(csv_path), } try: target_rows = fetch_target_rows(conn) updates, summary = build_plan(target_rows, source_rows) report.update(summary) report["sample_diffs"] = updates[: max(0, args.sample_limit)] report["focus_parent_ids"] = focus_parent_ids report["focus_parent_diffs"] = build_focus_summary(updates, focus_parent_ids) if args.apply: applied = apply_updates(conn, updates) conn.commit() report["applied_updates"] = applied else: conn.rollback() finally: conn.close() print(json.dumps(report, ensure_ascii=False, indent=2)) return 0 if __name__ == "__main__": raise SystemExit(main())