#!/usr/bin/env python3 from __future__ import annotations import argparse import base64 import hashlib import json import shutil import subprocess import urllib.error import urllib.parse import urllib.request import xml.etree.ElementTree as ET import zipfile from datetime import datetime, timezone from pathlib import Path from typing import Any SOAPENV_NS = "http://schemas.xmlsoap.org/soap/envelope/" METADATA_NS = "http://soap.sforce.com/2006/04/metadata" DEFAULT_LOGIN_DOMAIN = "login.salesforce.com" DEFAULT_API_VERSION = "63.0" DEFAULT_MANIFEST_NAME = "manifest_sharing_rules.json" DEFAULT_OUTPUT_SUBDIR = "sharingRules" DEFAULT_OBJECTS = ["Account", "Lead", "Opportunity", "Contact"] def utc_now() -> datetime: return datetime.now(timezone.utc) def ensure_dir(path: Path) -> None: path.mkdir(parents=True, exist_ok=True) def sha256_file(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as fh: for chunk in iter(lambda: fh.read(1024 * 1024), b""): digest.update(chunk) return digest.hexdigest() def load_kv_config(path: Path) -> dict[str, str]: data: dict[str, str] = {} for raw in path.read_text(encoding="utf-8").splitlines(): line = raw.strip() if not line or line.startswith("#") or ":" not in line: continue key, value = line.split(":", 1) data[key.strip()] = value.strip() return data def load_json(path: Path) -> dict[str, Any]: payload = json.loads(path.read_text(encoding="utf-8")) if not isinstance(payload, dict): raise ValueError(f"{path} must be a JSON object") return payload def xml_escape(value: str) -> str: return ( value.replace("&", "&") .replace("<", "<") .replace(">", ">") .replace('"', """) .replace("'", "'") ) class SalesforceMetadataClient: def __init__(self, config: dict[str, str], api_version: str) -> None: self.login_domain = (config.get("tokenDomain") or DEFAULT_LOGIN_DOMAIN).replace("https://", "").replace( "http://", "" ).strip("/") self.client_id = config["clientId"] self.client_secret = config["clientSecret"] self.username = config["username"] self.password = config["password"] self.api_version = api_version self.instance_url: str | None = None self.access_token: str | None = None def login(self) -> None: token_url = f"https://{self.login_domain}/services/oauth2/token" payload = urllib.parse.urlencode( { "grant_type": "password", "client_id": self.client_id, "client_secret": self.client_secret, "username": self.username, "password": self.password, } ).encode("utf-8") req = urllib.request.Request(token_url, data=payload, method="POST") with urllib.request.urlopen(req, timeout=30) as resp: data = json.loads(resp.read().decode("utf-8")) self.instance_url = data["instance_url"] self.access_token = data["access_token"] def _metadata_endpoint(self) -> str: if not self.instance_url or not self.access_token: self.login() assert self.instance_url is not None return f"{self.instance_url}/services/Soap/m/{self.api_version}" def _soap_request(self, body: str) -> ET.Element: if not self.instance_url or not self.access_token: self.login() assert self.access_token is not None req = urllib.request.Request( self._metadata_endpoint(), data=body.encode("utf-8"), method="POST", headers={ "Content-Type": "text/xml; charset=utf-8", "SOAPAction": '""', "Authorization": f"Bearer {self.access_token}", }, ) try: with urllib.request.urlopen(req, timeout=120) as resp: payload = resp.read().decode("utf-8") except urllib.error.HTTPError as err: payload = err.read().decode("utf-8", errors="replace") try: root = ET.fromstring(payload) fault = root.find(".//{%s}Fault" % SOAPENV_NS) if fault is not None: fault_code = fault.findtext("faultcode", default="").strip() fault_string = fault.findtext("faultstring", default="").strip() raise RuntimeError( f"Metadata API HTTP {err.code}: {fault_code or 'Fault'} {fault_string}".strip() ) from err except ET.ParseError: pass raise RuntimeError(f"Metadata API HTTP {err.code}: {payload}") from err root = ET.fromstring(payload) fault = root.find(".//{%s}Fault" % SOAPENV_NS) if fault is not None: fault_code = fault.findtext("faultcode", default="").strip() fault_string = fault.findtext("faultstring", default="").strip() raise RuntimeError(f"Metadata API fault {fault_code or ''} {fault_string or ''}".strip()) return root def retrieve_sharing_rules_zip(self, objects: list[str]) -> bytes: if not self.instance_url or not self.access_token: self.login() package_members = "".join(f"{xml_escape(obj)}" for obj in objects) root = self._soap_request( f""" {xml_escape(self.access_token or '')} {xml_escape(self.api_version)} true {package_members} SharingRules {xml_escape(self.api_version)} """ ) async_id = root.findtext(".//{%s}id" % METADATA_NS, default="").strip() if not async_id: raise RuntimeError("retrieve() did not return an async id") while True: status_root = self._soap_request( f""" {xml_escape(self.access_token or '')} {xml_escape(async_id)} true """ ) done = status_root.findtext(".//{%s}done" % METADATA_NS, default="false").strip().lower() == "true" state = status_root.findtext(".//{%s}status" % METADATA_NS, default="").strip() if state in {"Failed", "Canceled"}: message = status_root.findtext(".//{%s}errorMessage" % METADATA_NS, default="").strip() raise RuntimeError(f"retrieve status={state}: {message}") if done: zip_file_b64 = status_root.findtext(".//{%s}zipFile" % METADATA_NS, default="").strip() if not zip_file_b64: raise RuntimeError("checkRetrieveStatus completed without zipFile") return base64.b64decode(zip_file_b64) def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description="Export SharingRules metadata into the standard snapshot directory.") parser.add_argument("--config", required=True, help="Path to Salesforce config file.") parser.add_argument("--snapshot-dir", required=True, help="Existing daily snapshot directory.") parser.add_argument("--api-version", default=DEFAULT_API_VERSION, help="Salesforce Metadata API version.") parser.add_argument("--base-manifest-name", default="manifest.json", help="Base snapshot manifest name.") parser.add_argument("--manifest-name", default=DEFAULT_MANIFEST_NAME, help="Manifest file name inside snapshot-dir.") parser.add_argument("--output-subdir", default=DEFAULT_OUTPUT_SUBDIR, help="Directory under snapshot-dir for extracted XML.") parser.add_argument("--zip-name", default="sharingRules-retrieve.zip", help="Zip artifact file name inside snapshot-dir.") parser.add_argument("--s3-bucket", help="Optional S3 bucket override.") parser.add_argument("--s3-prefix", help="Optional S3 prefix override.") parser.add_argument("--skip-s3-upload", action="store_true", help="Skip S3 upload.") parser.add_argument("--object", dest="objects", action="append", default=[], help="SharingRules member name. Repeatable.") return parser.parse_args() def upload_file_to_s3(local_path: Path, s3_uri: str) -> None: aws_path = shutil.which("aws") if not aws_path: raise RuntimeError("aws CLI is required for S3 upload but was not found in PATH") subprocess.run([aws_path, "s3", "cp", str(local_path), s3_uri], check=True) def resolve_s3_target(snapshot_dir: Path, args: argparse.Namespace, base_manifest: dict[str, Any]) -> str | None: if args.skip_s3_upload: return None if args.s3_bucket: prefix = (args.s3_prefix or "salesforce/full").strip("/") parts = snapshot_dir.parts[-2:] return f"s3://{args.s3_bucket}/{prefix}/{parts[0]}/{parts[1]}/" s3_uri = base_manifest.get("s3_uri") return str(s3_uri) if s3_uri else None def extract_requested_files(zip_path: Path, output_dir: Path, objects: list[str]) -> list[dict[str, Any]]: extracted: list[dict[str, Any]] = [] with zipfile.ZipFile(zip_path) as zf: names = zf.namelist() for object_name in objects: suffixes = [ f"sharingRules/{object_name}.sharingRules-meta.xml", f"sharingRules/{object_name}.sharingRules", ] member = next((name for name in names if any(name.endswith(suffix) for suffix in suffixes)), None) if not member: continue target_path = output_dir / f"{object_name}.sharingRules-meta.xml" target_path.write_bytes(zf.read(member)) extracted.append( { "object": object_name, "file": str(target_path.relative_to(output_dir.parent)), "sha256": sha256_file(target_path), } ) return extracted def main() -> int: args = parse_args() snapshot_dir = Path(args.snapshot_dir).expanduser().resolve() ensure_dir(snapshot_dir) output_dir = snapshot_dir / args.output_subdir ensure_dir(output_dir) base_manifest_path = snapshot_dir / args.base_manifest_name base_manifest = load_json(base_manifest_path) if base_manifest_path.exists() else {} config = load_kv_config(Path(args.config).expanduser().resolve()) client = SalesforceMetadataClient(config, args.api_version) objects = args.objects or list(DEFAULT_OBJECTS) zip_bytes = client.retrieve_sharing_rules_zip(objects) zip_path = snapshot_dir / args.zip_name zip_path.write_bytes(zip_bytes) extracted = extract_requested_files(zip_path, output_dir, objects) manifest = { "manifest_version": 1, "object": "SharingRules", "export_type": "sharing_rules_metadata", "snapshot_time": utc_now().isoformat(), "snapshot_dir": str(snapshot_dir), "source_org": client.instance_url, "api_version": args.api_version, "objects_requested": objects, "objects_extracted": [item["object"] for item in extracted], "zip_file": zip_path.name, "zip_sha256": sha256_file(zip_path), "output_subdir": output_dir.name, "files": extracted, "base_manifest": base_manifest_path.name if base_manifest_path.exists() else None, } manifest_path = snapshot_dir / args.manifest_name manifest_path.write_text(json.dumps(manifest, indent=2), encoding="utf-8") s3_uri = resolve_s3_target(snapshot_dir, args, base_manifest) if s3_uri: zip_uri = s3_uri.rstrip("/") + "/" + zip_path.name upload_file_to_s3(zip_path, zip_uri) for item in extracted: upload_file_to_s3(snapshot_dir / item["file"], s3_uri.rstrip("/") + "/" + item["file"]) manifest_uri = s3_uri.rstrip("/") + "/" + manifest_path.name manifest["s3_uri"] = manifest_uri manifest_path.write_text(json.dumps(manifest, indent=2), encoding="utf-8") upload_file_to_s3(manifest_path, manifest_uri) print(f"[ok] manifest={manifest_path}", flush=True) print(f"[ok] zip={zip_path}", flush=True) return 0 if __name__ == "__main__": raise SystemExit(main())