#!/usr/bin/env python3 """Apply the one L2-owned SCIM policy singleton through the Seed dispatcher. The native SCIM surface reads ``IdentitySyncSetting(Name=default)`` under system context and deliberately has no public configuration write route. This adapter is therefore the only production writer: it provisions and proves the dedicated zero-access first-onboarding Profile through the audited runtime API, resolves the currently active break-glass administrators, writes the single policy row idempotently, and proves the persisted policy before reporting success. It never prints principal values or database credentials. """ from __future__ import annotations import argparse import json import os from pathlib import Path import shutil import subprocess import sys import time from typing import Any SCRIPT_DIR = Path(__file__).resolve().parent REPO_ROOT = SCRIPT_DIR.parent RUNTIME_DIR = REPO_ROOT / "deployments" / "offline" / "runtime" if str(RUNTIME_DIR) not in sys.path: sys.path.insert(0, str(RUNTIME_DIR)) import deploy_offline_env # noqa: E402 MODULE_ID = "identity_sync_policy" SUMMARY_FILENAME = "summary.json" SUMMARY_FORMAT = "deploy-seed-script-v1" PLANNED_COUNTS = { "identity_sync_settings": 1, "onboarding_profiles": 1, "service_profiles": 1, "service_users": 1, } SEED_RELATIVE_PATH = Path("packs/Salesforce-PingCAP-L2/identity-sync-policy.seed.json") REQUIRED_SINGLETON_FIELDS = { "name", "defaultProfileName", "serviceProfileName", "serviceUsername", "protectedProfileNames", "breakerFailureThreshold", "breakerCooldownSeconds", "tokenTtlHours", "timeZoneSidKey", "localeSidKey", "emailEncodingKey", "languageLocaleKey", } REQUIRED_ONBOARDING_PROFILE_FIELDS = { "description", "isActive", "isSystem", "userType", } REQUIRED_SERVICE_PROFILE_FIELDS = { "description", "isActive", "isSystem", "userType", "systemPermissions", "objectPermissions", "fieldPermissions", "denyUnlistedFieldsForObjects", } REQUIRED_SERVICE_USER_FIELDS = { "lastName", "alias", "isActive", "userRoleId", } SERVICE_PROFILE_SYSTEM_PERMISSIONS = ["ManageUsers"] SERVICE_PROFILE_OBJECT_PERMISSIONS = [ { "object": "IdentitySyncDeactivationTicket", "read": True, "create": True, "edit": False, "delete": False, "viewAllRecords": False, "modifyAllRecords": False, }, { "object": "IdentitySyncSetting", "read": True, "create": False, "edit": False, "delete": False, "viewAllRecords": True, "modifyAllRecords": False, }, { "object": "Profile", "read": True, "create": False, "edit": False, "delete": False, "viewAllRecords": False, "modifyAllRecords": False, }, { "object": "User", "read": True, "create": True, "edit": True, "delete": False, "viewAllRecords": False, "modifyAllRecords": False, }, ] SERVICE_PROFILE_FIELD_PERMISSIONS = [ { "object": "IdentitySyncDeactivationTicket", "fields": ["Bucket", "Name", "Slot"], "read": True, "edit": True, }, { "object": "IdentitySyncSetting", "fields": [ "BreakerCooldownSeconds", "BreakerFailureThreshold", "DefaultProfileName", "EmailEncodingKey", "ExemptPrincipals", "LanguageLocaleKey", "LocaleSidKey", "Name", "ServiceProfileName", "ServiceUsername", "TimeZoneSidKey", "TokenTtlHours", ], "read": True, "edit": False, }, { "object": "Profile", "fields": ["Id", "Name"], "read": True, "edit": False, }, { "object": "User", "fields": ["Id"], "read": True, "edit": False, }, { "object": "User", "fields": [ "Alias", "Department", "Email", "EmailEncodingKey", "FirstName", "IsActive", "LanguageLocaleKey", "LastName", "LocaleSidKey", "MobilePhone", "Phone", "ProfileId", "TimeZoneSidKey", "Title", "UserRoleId", "Username", ], "read": True, "edit": True, }, ] SERVICE_PROFILE_DENY_UNLISTED_FIELDS_FOR_OBJECTS = [ "IdentitySyncDeactivationTicket", "IdentitySyncSetting", "Profile", "User", ] class IdentitySyncPolicySeedError(RuntimeError): """Raised when the L2 SCIM policy cannot be safely applied.""" def parse_args(argv: list[str] | None = None) -> argparse.Namespace: parser = argparse.ArgumentParser(description="Apply the L2 SCIM policy singleton.") parser.add_argument("--source-root", required=True) parser.add_argument("--module-id", required=True) parser.add_argument("--invocation-id", required=True) parser.add_argument("--data-source", action="append", required=True) parser.add_argument("--selection-json", required=True) parser.add_argument("--output-dir", required=True) parser.add_argument("--dry-run", action="store_true") return parser.parse_args(argv) def sql_literal(value: str | int) -> str: if isinstance(value, int): return str(value) return "'" + value.replace("\\", "\\\\").replace("'", "''") + "'" def mysql_base_args() -> list[str]: settings = { "TIDB_HOST": os.environ.get("TIDB_HOST", "").strip(), "TIDB_PORT": os.environ.get("TIDB_PORT", "").strip(), "TIDB_USER": os.environ.get("TIDB_USER", "").strip(), "TIDB_PASSWORD": os.environ.get("TIDB_PASSWORD", ""), "TIDB_DATABASE": os.environ.get("TIDB_DATABASE", "").strip(), } missing = [name for name, value in settings.items() if not value] if missing: raise IdentitySyncPolicySeedError("missing database settings: " + ", ".join(missing)) if not settings["TIDB_PORT"].isdigit(): raise IdentitySyncPolicySeedError("TIDB_PORT must be numeric") client = os.environ.get("IDENTITY_SYNC_POLICY_MYSQL_CLIENT", "").strip() if not client: client = shutil.which("mariadb") or shutil.which("mysql") or "" if not client: raise IdentitySyncPolicySeedError("missing required command: mysql or mariadb") command = [ client, "--batch", "--raw", "--skip-column-names", "--host", settings["TIDB_HOST"], "--port", settings["TIDB_PORT"], "--user", settings["TIDB_USER"], settings["TIDB_DATABASE"], ] if os.environ.get("TIDB_TLS", "true").strip().lower() in {"", "1", "true", "yes", "on"}: command.insert(1, "--ssl") return command def mysql_query(sql: str) -> str: command = mysql_base_args() env = { "PATH": os.environ.get("PATH") or os.defpath, "MYSQL_PWD": os.environ.get("TIDB_PASSWORD", ""), } try: result = subprocess.run( [*command, "-e", sql], check=True, capture_output=True, text=True, env=env, ) except (OSError, subprocess.CalledProcessError) as exc: raise IdentitySyncPolicySeedError("SCIM L2 policy database operation failed") from exc return result.stdout def load_seed(source_root: Path, data_sources: list[str]) -> dict[str, Any]: if len(data_sources) != 1: raise IdentitySyncPolicySeedError("SCIM policy Seed requires exactly one data source") expected = (source_root / SEED_RELATIVE_PATH).resolve(strict=True) actual = Path(data_sources[0]).expanduser().resolve(strict=True) if actual != expected: raise IdentitySyncPolicySeedError("data source is not the reviewed SCIM policy Seed") try: payload = json.loads(actual.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError) as exc: raise IdentitySyncPolicySeedError("SCIM policy Seed is unreadable") from exc if not isinstance(payload, dict) or set(payload) != { "version", "onboardingProfile", "serviceProfile", "serviceUser", "singleton", }: raise IdentitySyncPolicySeedError("SCIM policy Seed has an unsupported shape") if ( payload["version"] != 3 or not isinstance(payload["singleton"], dict) or not isinstance(payload["onboardingProfile"], dict) or not isinstance(payload["serviceProfile"], dict) or not isinstance(payload["serviceUser"], dict) ): raise IdentitySyncPolicySeedError("SCIM policy Seed has an unsupported version") singleton = payload["singleton"] if set(singleton) != REQUIRED_SINGLETON_FIELDS: raise IdentitySyncPolicySeedError( "SCIM policy singleton fields do not match the reviewed contract" ) integer_fields = { "breakerFailureThreshold", "breakerCooldownSeconds", "tokenTtlHours", } for field in REQUIRED_SINGLETON_FIELDS - {"protectedProfileNames", *integer_fields}: if not isinstance(singleton[field], str) or not singleton[field].strip(): raise IdentitySyncPolicySeedError(f"SCIM policy {field} must be a non-empty string") if singleton["name"] != "default": raise IdentitySyncPolicySeedError("SCIM policy singleton name must be default") if singleton["defaultProfileName"] == singleton["serviceProfileName"]: raise IdentitySyncPolicySeedError("SCIM default and service profiles must be distinct") if not singleton["serviceUsername"].lower().endswith(".invalid"): raise IdentitySyncPolicySeedError( "SCIM service username must use the reserved .invalid domain" ) protected_profiles = singleton["protectedProfileNames"] if ( not isinstance(protected_profiles, list) or not protected_profiles or len(protected_profiles) != len(set(protected_profiles)) or any(not isinstance(name, str) or not name.strip() for name in protected_profiles) ): raise IdentitySyncPolicySeedError( "SCIM protected profiles must be a non-empty unique string list" ) if singleton["defaultProfileName"] in protected_profiles: raise IdentitySyncPolicySeedError( "SCIM onboarding profile must not be protected or administrative" ) for field in integer_fields: if ( isinstance(singleton[field], bool) or not isinstance(singleton[field], int) or singleton[field] <= 0 ): raise IdentitySyncPolicySeedError(f"SCIM policy {field} must be a positive integer") onboarding_profile = payload["onboardingProfile"] if set(onboarding_profile) != REQUIRED_ONBOARDING_PROFILE_FIELDS: raise IdentitySyncPolicySeedError( "SCIM onboarding profile fields do not match the reviewed contract" ) if ( not isinstance(onboarding_profile["description"], str) or not onboarding_profile["description"].strip() ): raise IdentitySyncPolicySeedError( "SCIM onboarding profile description must be a non-empty string" ) if onboarding_profile["isActive"] is not True: raise IdentitySyncPolicySeedError("SCIM onboarding profile must be active") if onboarding_profile["isSystem"] is not False: raise IdentitySyncPolicySeedError("SCIM onboarding profile must be non-system") if onboarding_profile["userType"] != "Standard": raise IdentitySyncPolicySeedError("SCIM onboarding profile must use Standard user type") service_profile = payload["serviceProfile"] if set(service_profile) != REQUIRED_SERVICE_PROFILE_FIELDS: raise IdentitySyncPolicySeedError( "SCIM service profile fields do not match the reviewed contract" ) if ( not isinstance(service_profile["description"], str) or not service_profile["description"].strip() or service_profile["isActive"] is not True or service_profile["isSystem"] is not False or service_profile["userType"] != "Standard" or service_profile["systemPermissions"] != SERVICE_PROFILE_SYSTEM_PERMISSIONS or service_profile["objectPermissions"] != SERVICE_PROFILE_OBJECT_PERMISSIONS or service_profile["fieldPermissions"] != SERVICE_PROFILE_FIELD_PERMISSIONS or service_profile["denyUnlistedFieldsForObjects"] != SERVICE_PROFILE_DENY_UNLISTED_FIELDS_FOR_OBJECTS ): raise IdentitySyncPolicySeedError( "SCIM service profile does not match the reviewed native SCIM permission matrix" ) service_user = payload["serviceUser"] if set(service_user) != REQUIRED_SERVICE_USER_FIELDS or service_user != { "lastName": "SCIM Provisioning", "alias": "scimprov", "isActive": True, "userRoleId": None, }: raise IdentitySyncPolicySeedError( "SCIM service user does not match the reviewed native SCIM principal contract" ) return { "singleton": singleton, "onboardingProfile": onboarding_profile, "serviceProfile": service_profile, "serviceUser": service_user, } def runtime_base_url() -> str: port = os.environ.get("BACKEND_PORT", "").strip() if not port.isdigit(): raise IdentitySyncPolicySeedError( "BACKEND_PORT must be numeric for SCIM onboarding profile provisioning" ) return f"http://127.0.0.1:{port}" def runtime_admin_access_token() -> str: try: return deploy_offline_env.request_exp_local_access_token(dict(os.environ)) except (deploy_offline_env.DeployError, KeyError, OSError, ValueError) as exc: raise IdentitySyncPolicySeedError( "SCIM onboarding profile admin authentication failed" ) from exc def _tab_rows(raw: str) -> list[list[str]]: return [line.split("\t") for line in raw.rstrip("\n").splitlines() if line] def _onboarding_profile_id(profile_name: str) -> str | None: rows = _tab_rows( mysql_query( "SELECT Id, IsActive, IsSystem, UserType, IsDeleted FROM Profile WHERE Name = " + sql_literal(profile_name) + " ORDER BY Id" ) ) if not rows: return None if len(rows) != 1 or len(rows[0]) != 5: raise IdentitySyncPolicySeedError( "SCIM onboarding profile name is duplicate or its readback is malformed" ) profile_id, is_active, is_system, user_type, is_deleted = rows[0] if ( not profile_id or is_active != "1" or is_system != "0" or user_type != "Standard" or is_deleted != "0" ): raise IdentitySyncPolicySeedError( "SCIM onboarding profile does not match the active non-system Standard contract" ) return profile_id def _service_profile_id(profile_name: str) -> str | None: rows = _tab_rows( mysql_query( "SELECT Id, IsActive, IsSystem, UserType, IsDeleted FROM Profile WHERE Name = " + sql_literal(profile_name) + " ORDER BY Id" ) ) if not rows: return None if len(rows) != 1 or len(rows[0]) != 5: raise IdentitySyncPolicySeedError( "SCIM service profile name is duplicate or its readback is malformed" ) profile_id, is_active, is_system, user_type, is_deleted = rows[0] if ( not profile_id or is_active != "1" or is_system != "0" or user_type != "Standard" or is_deleted != "0" ): raise IdentitySyncPolicySeedError( "SCIM service profile does not match the active non-system Standard contract" ) return profile_id def _profile_permission_set_id(profile_id: str) -> str: rows = _tab_rows( mysql_query( "SELECT Id FROM PermissionSet WHERE ProfileId = " + sql_literal(profile_id) + " AND IsOwnedByProfile = 1 AND IsActive = 1 AND IsDeleted = 0 ORDER BY Id" ) ) if len(rows) != 1 or len(rows[0]) != 1 or not rows[0][0]: raise IdentitySyncPolicySeedError( "SCIM profile must have exactly one active profile-owned PermissionSet" ) return rows[0][0] def zero_access_count_queries(profile_id: str, permission_set_id: str) -> list[tuple[str, str]]: profile = sql_literal(profile_id) permission_set = sql_literal(permission_set_id) return [ ( "profile system permissions", "SELECT COUNT(*) FROM _System_ProfilePermission WHERE ProfileId = " + profile + " AND IsEnabled = 1", ), ( "enabled system permissions", "SELECT COUNT(*) FROM _System_PermissionSetPermission WHERE PermissionSetId = " + permission_set + " AND IsEnabled = 1", ), ( "object permissions", "SELECT COUNT(*) FROM ObjectPermissions WHERE ParentId = " + permission_set + " AND COALESCE(IsDeleted, 0) = 0 AND (COALESCE(PermissionsRead, 0) = 1 OR " "COALESCE(PermissionsCreate, 0) = 1 OR COALESCE(PermissionsEdit, 0) = 1 OR " "COALESCE(PermissionsDelete, 0) = 1 OR COALESCE(PermissionsViewAllRecords, 0) = 1 OR " "COALESCE(PermissionsModifyAllRecords, 0) = 1)", ), ( "field permissions", "SELECT COUNT(*) FROM FieldPermissions WHERE ParentId = " + permission_set + " AND COALESCE(IsDeleted, 0) = 0 AND (COALESCE(PermissionsRead, 0) = 1 OR " "COALESCE(PermissionsEdit, 0) = 1)", ), ( "permission-set custom permissions", "SELECT COUNT(*) FROM PermissionSetCustomPermission WHERE PermissionSetId = " + permission_set, ), ( "profile custom permissions", "SELECT COUNT(*) FROM ProfileCustomPermission WHERE ProfileId = " + profile, ), ( "record-type visibility", "SELECT COUNT(*) FROM PermissionSetRecordTypeVisibility WHERE PermissionSetId = " + permission_set + " AND IsVisible = 1", ), ] def verify_zero_access_profile(profile_id: str) -> None: permission_set_id = _profile_permission_set_id(profile_id) for label, query in zero_access_count_queries(profile_id, permission_set_id): count = mysql_query(query).strip() if count != "0": raise IdentitySyncPolicySeedError( f"SCIM onboarding profile is not zero-access: {label} are present" ) def ensure_onboarding_profile( access_token: str, singleton: dict[str, Any], onboarding_profile: dict[str, Any], ) -> str: profile_name = singleton["defaultProfileName"] profile_id = _onboarding_profile_id(profile_name) if profile_id is None: try: response = deploy_offline_env.api_json_request( runtime_base_url(), "/services/data/v63.0/sobjects/Profile", access_token=access_token, method="POST", payload={ "Name": profile_name, "Description": onboarding_profile["description"], }, ) except (deploy_offline_env.DeployError, OSError, ValueError) as exc: raise IdentitySyncPolicySeedError( "SCIM onboarding profile creation failed" ) from exc created_profile_id = ( str(response.get("id") or response.get("Id") or "").strip() if isinstance(response, dict) else "" ) if not created_profile_id: raise IdentitySyncPolicySeedError( "SCIM onboarding profile creation returned no id" ) profile_id = _onboarding_profile_id(profile_name) if profile_id is None: raise IdentitySyncPolicySeedError( "SCIM onboarding profile was not persisted" ) verify_zero_access_profile(profile_id) return profile_id def _service_object_permission_snapshot(permission_set_id: str) -> dict[str, tuple[str, ...]]: rows = _tab_rows( mysql_query( "SELECT Id, SobjectType, PermissionsRead, PermissionsCreate, PermissionsEdit, " "PermissionsDelete, PermissionsViewAllRecords, PermissionsModifyAllRecords " "FROM ObjectPermissions WHERE ParentId = " + sql_literal(permission_set_id) + " AND COALESCE(IsDeleted, 0) = 0 ORDER BY SobjectType, Id" ) ) result: dict[str, tuple[str, ...]] = {} for row in rows: if ( len(row) != 8 or not row[0] or not row[1] or any(value not in {"0", "1"} for value in row[2:]) ): raise IdentitySyncPolicySeedError( "SCIM service profile object-permission readback is malformed" ) if row[1] in result: raise IdentitySyncPolicySeedError( "SCIM service profile has duplicate object permissions" ) result[row[1]] = tuple(row) return result def _expected_service_object_permissions() -> dict[str, tuple[str, ...]]: return { permission["object"]: tuple( "1" if permission[field] else "0" for field in ( "read", "create", "edit", "delete", "viewAllRecords", "modifyAllRecords", ) ) for permission in SERVICE_PROFILE_OBJECT_PERMISSIONS } def _service_field_permission_snapshot( permission_set_id: str, ) -> dict[tuple[str, str], tuple[str, ...]]: rows = _tab_rows( mysql_query( "SELECT Id, SobjectType, Field, PermissionsRead, PermissionsEdit " "FROM FieldPermissions WHERE ParentId = " + sql_literal(permission_set_id) + " AND COALESCE(IsDeleted, 0) = 0 ORDER BY SobjectType, Field, Id" ) ) result: dict[tuple[str, str], tuple[str, ...]] = {} for row in rows: if ( len(row) != 5 or not row[0] or not row[1] or not row[2] or any(value not in {"0", "1"} for value in row[3:]) ): raise IdentitySyncPolicySeedError( "SCIM service profile field-permission readback is malformed" ) field_name = row[2].removeprefix(row[1] + ".") key = (row[1], field_name) if not field_name or key in result: raise IdentitySyncPolicySeedError( "SCIM service profile has duplicate field permissions" ) result[key] = tuple(row) return result def _service_profile_field_catalog() -> set[tuple[str, str]]: covered_objects = SERVICE_PROFILE_DENY_UNLISTED_FIELDS_FOR_OBJECTS rows = _tab_rows( mysql_query( "SELECT o.ApiName, f.ApiName FROM _System_Field AS f " "INNER JOIN _System_Object AS o ON o.Id = f.ObjectId " "WHERE COALESCE(f.IsDeleted, 0) = 0 AND COALESCE(o.IsDeleted, 0) = 0 " "AND o.ApiName IN (" + ", ".join(sql_literal(name) for name in covered_objects) + ") ORDER BY o.ApiName, f.ApiName" ) ) result: set[tuple[str, str]] = set() for row in rows: if len(row) != 2 or not row[0] or not row[1]: raise IdentitySyncPolicySeedError( "SCIM service profile field catalog readback is malformed" ) result.add((row[0], row[1])) observed_objects = {object_name for object_name, _ in result} if observed_objects != set(covered_objects): raise IdentitySyncPolicySeedError( "SCIM service profile field catalog is incomplete" ) return result def _expected_service_field_permissions( field_catalog: set[tuple[str, str]], ) -> dict[tuple[str, str], tuple[str, ...]]: configured: dict[tuple[str, str], tuple[str, ...]] = {} for permission in SERVICE_PROFILE_FIELD_PERMISSIONS: values = ( "1" if permission["read"] else "0", "1" if permission["edit"] else "0", ) for field in permission["fields"]: configured[(permission["object"], field)] = values if configured.keys() - field_catalog: raise IdentitySyncPolicySeedError( "SCIM service profile references fields missing from runtime metadata" ) return {key: configured.get(key, ("0", "0")) for key in field_catalog} def verify_service_profile_has_no_auxiliary_grants( profile_id: str, permission_set_id: str, ) -> None: profile = sql_literal(profile_id) permission_set = sql_literal(permission_set_id) count_queries = ( "SELECT COUNT(*) FROM _System_ProfilePermission WHERE ProfileId = " + profile + " AND IsEnabled = 1", "SELECT COUNT(*) FROM ProfileCustomPermission WHERE ProfileId = " + profile, "SELECT COUNT(*) FROM PermissionSetCustomPermission WHERE PermissionSetId = " + permission_set, "SELECT COUNT(*) FROM PermissionSetRecordTypeVisibility WHERE PermissionSetId = " + permission_set + " AND IsVisible = 1", ) if [mysql_query(query).strip() for query in count_queries] != ["0"] * len( count_queries ): raise IdentitySyncPolicySeedError( "SCIM service profile has unreviewed auxiliary permission grants" ) def verify_service_profile_permissions( profile_id: str, permission_set_id: str, ) -> None: verify_service_profile_has_no_auxiliary_grants(profile_id, permission_set_id) system_permissions = { value for value in mysql_query( "SELECT PermissionApiName FROM _System_PermissionSetPermission WHERE PermissionSetId = " + sql_literal(permission_set_id) + " AND IsEnabled = 1 ORDER BY PermissionApiName" ).splitlines() if value } if system_permissions != set(SERVICE_PROFILE_SYSTEM_PERMISSIONS): raise IdentitySyncPolicySeedError( "SCIM service profile system permissions do not match the reviewed contract" ) object_snapshot = _service_object_permission_snapshot(permission_set_id) expected_objects = _expected_service_object_permissions() enabled_objects = { object_name: row[2:] for object_name, row in object_snapshot.items() if "1" in row[2:] } if enabled_objects != expected_objects: raise IdentitySyncPolicySeedError( "SCIM service profile object permissions do not match the reviewed contract" ) field_snapshot = _service_field_permission_snapshot(permission_set_id) expected_fields = _expected_service_field_permissions( _service_profile_field_catalog() ) extra_enabled_fields = { key: row[3:] for key, row in field_snapshot.items() if key not in expected_fields and "1" in row[3:] } if extra_enabled_fields or any( key not in field_snapshot or field_snapshot[key][3:] != values for key, values in expected_fields.items() ): raise IdentitySyncPolicySeedError( "SCIM service profile field permissions do not match the reviewed contract" ) def _service_profile_api_request( access_token: str, path: str, *, method: str, payload: dict[str, Any], ) -> dict[str, Any]: try: response = deploy_offline_env.api_json_request( runtime_base_url(), path, access_token=access_token, method=method, payload=payload, ) except (deploy_offline_env.DeployError, OSError, ValueError) as exc: raise IdentitySyncPolicySeedError( "SCIM service profile permission reconciliation failed" ) from exc return response if isinstance(response, dict) else {} def reconcile_service_profile_permissions( access_token: str, profile_id: str, permission_set_id: str, ) -> None: verify_service_profile_has_no_auxiliary_grants(profile_id, permission_set_id) expected_fields = _expected_service_field_permissions( _service_profile_field_catalog() ) current_system_permissions = { value for value in mysql_query( "SELECT PermissionApiName FROM _System_PermissionSetPermission WHERE PermissionSetId = " + sql_literal(permission_set_id) + " AND IsEnabled = 1 ORDER BY PermissionApiName" ).splitlines() if value } expected_system_permissions = set(SERVICE_PROFILE_SYSTEM_PERMISSIONS) if current_system_permissions - expected_system_permissions: raise IdentitySyncPolicySeedError( "SCIM service profile has unreviewed enabled system permissions" ) if current_system_permissions != expected_system_permissions: _service_profile_api_request( access_token, "/services/data/v63.0/permissions/system/permissionset/" + permission_set_id, method="PATCH", payload={"grants": {name: True for name in SERVICE_PROFILE_SYSTEM_PERMISSIONS}}, ) object_snapshot = _service_object_permission_snapshot(permission_set_id) expected_objects = _expected_service_object_permissions() extra_enabled_objects = sorted( object_name for object_name, row in object_snapshot.items() if object_name not in expected_objects and "1" in row[2:] ) if extra_enabled_objects: raise IdentitySyncPolicySeedError( "SCIM service profile has unreviewed enabled object permissions" ) object_field_names = ( "PermissionsRead", "PermissionsCreate", "PermissionsEdit", "PermissionsDelete", "PermissionsViewAllRecords", "PermissionsModifyAllRecords", ) for object_name, expected_values in expected_objects.items(): payload = { field: value == "1" for field, value in zip(object_field_names, expected_values, strict=True) } current = object_snapshot.get(object_name) if current is None: _service_profile_api_request( access_token, "/services/data/v63.0/sobjects/ObjectPermissions", method="POST", payload={ "ParentId": permission_set_id, "SobjectType": object_name, **payload, }, ) elif current[2:] != expected_values: _service_profile_api_request( access_token, "/services/data/v63.0/sobjects/ObjectPermissions/" + current[0], method="PATCH", payload=payload, ) field_snapshot = _service_field_permission_snapshot(permission_set_id) extra_enabled_fields = sorted( key for key, row in field_snapshot.items() if key not in expected_fields and "1" in row[3:] ) if extra_enabled_fields: raise IdentitySyncPolicySeedError( "SCIM service profile has unreviewed enabled field permissions" ) for (object_name, field_name), expected_values in expected_fields.items(): payload = { "PermissionsRead": expected_values[0] == "1", "PermissionsEdit": expected_values[1] == "1", } current = field_snapshot.get((object_name, field_name)) if current is None: _service_profile_api_request( access_token, "/services/data/v63.0/sobjects/FieldPermissions", method="POST", payload={ "ParentId": permission_set_id, "SobjectType": object_name, "Field": object_name + "." + field_name, **payload, }, ) elif current[3:] != expected_values: _service_profile_api_request( access_token, "/services/data/v63.0/sobjects/FieldPermissions/" + current[0], method="PATCH", payload=payload, ) def ensure_service_profile( access_token: str, singleton: dict[str, Any], service_profile: dict[str, Any], ) -> str: profile_id = _service_profile_id(singleton["serviceProfileName"]) if profile_id is None: response = _service_profile_api_request( access_token, "/services/data/v63.0/sobjects/Profile", method="POST", payload={ "Name": singleton["serviceProfileName"], "Description": service_profile["description"], }, ) created_profile_id = str(response.get("id") or response.get("Id") or "").strip() if not created_profile_id: raise IdentitySyncPolicySeedError( "SCIM service profile creation returned no id" ) profile_id = _service_profile_id(singleton["serviceProfileName"]) if profile_id is None: raise IdentitySyncPolicySeedError("SCIM service profile was not persisted") permission_set_id = _profile_permission_set_id(profile_id) reconcile_service_profile_permissions(access_token, profile_id, permission_set_id) verify_service_profile_permissions(profile_id, permission_set_id) return profile_id def _service_user_row( singleton: dict[str, Any], ) -> tuple[str, str, str, str, str] | None: rows = _tab_rows( mysql_query( "SELECT Id, ProfileId, IsActive, UserRoleId, Email FROM `User` " "WHERE Username = " + sql_literal(singleton["serviceUsername"]) + " AND COALESCE(IsDeleted, 0) = 0 ORDER BY Id" ) ) if not rows: return None if ( len(rows) != 1 or len(rows[0]) != 5 or not rows[0][0] or not rows[0][1] or rows[0][2] not in {"0", "1"} ): raise IdentitySyncPolicySeedError("SCIM service user readback is malformed or ambiguous") user_id, profile_id, is_active, user_role_id, email = rows[0] if user_role_id == "NULL": user_role_id = "" if email == "NULL": email = "" else: email = email.strip().lower() return user_id, profile_id, is_active, user_role_id, email def verify_service_user_has_no_additional_grants(user_id: str) -> None: count_queries = ( "SELECT COUNT(*) FROM PermissionSetAssignment WHERE AssigneeId = " + sql_literal(user_id) + " AND COALESCE(IsDeleted, 0) = 0", "SELECT COUNT(*) FROM PermissionSetGroupAssignment WHERE AssigneeId = " + sql_literal(user_id) + " AND COALESCE(IsDeleted, 0) = 0", "SELECT COUNT(*) FROM GroupMember WHERE UserOrGroupId = " + sql_literal(user_id) + " AND COALESCE(IsDeleted, 0) = 0", ) counts = [mysql_query(query).strip() for query in count_queries] if counts != ["0", "0", "0"]: raise IdentitySyncPolicySeedError( "SCIM service user has additional permission assignments or group memberships" ) def ensure_service_user( access_token: str, singleton: dict[str, Any], service_user: dict[str, Any], profile_id: str, ) -> None: current = _service_user_row(singleton) if current is None: response = _service_profile_api_request( access_token, "/services/data/v63.0/sobjects/User", method="POST", payload={ "Username": singleton["serviceUsername"], "Email": singleton["serviceUsername"], "LastName": service_user["lastName"], "Alias": service_user["alias"], "ProfileId": profile_id, "IsActive": service_user["isActive"], "UserRoleId": service_user["userRoleId"], }, ) user_id = str(response.get("id") or response.get("Id") or "").strip() if not user_id: raise IdentitySyncPolicySeedError( "SCIM service user creation returned no id" ) else: user_id, current_profile_id, is_active, user_role_id, email = current verify_service_user_has_no_additional_grants(user_id) updates: dict[str, Any] = {} if current_profile_id != profile_id: updates["ProfileId"] = profile_id if is_active != "1": updates["IsActive"] = True if user_role_id: updates["UserRoleId"] = None if not strings_equal_fold(email, singleton["serviceUsername"]): updates["Email"] = singleton["serviceUsername"] if updates: _service_profile_api_request( access_token, "/services/data/v63.0/sobjects/User/" + user_id, method="PATCH", payload=updates, ) readback = _service_user_row(singleton) expected = ( user_id, profile_id, "1", "", singleton["serviceUsername"].strip().lower(), ) if readback != expected: raise IdentitySyncPolicySeedError( "SCIM service user readback did not match the L2 Seed" ) verify_service_user_has_no_additional_grants(user_id) def strings_equal_fold(left: str, right: str) -> bool: return left.strip().lower() == right.strip().lower() def require_profiles( singleton: dict[str, Any], *, include_onboarding: bool = True, include_service: bool = True, ) -> None: names = [*singleton["protectedProfileNames"]] if include_onboarding: names.append(singleton["defaultProfileName"]) if include_service: names.append(singleton["serviceProfileName"]) unique_names = sorted(set(names)) rows = { line.strip() for line in mysql_query( "SELECT Name FROM Profile WHERE Name IN (" + ", ".join(sql_literal(name) for name in unique_names) + ")" ).splitlines() if line.strip() } missing = sorted(set(unique_names) - rows) if missing: raise IdentitySyncPolicySeedError("SCIM policy references missing configured profile names") def protected_principals(singleton: dict[str, Any]) -> list[str]: profile_names = singleton["protectedProfileNames"] profile_filter = ", ".join(sql_literal(name) for name in profile_names) query = f""" SELECT DISTINCT LOWER(TRIM(principal)) AS normalized_principal FROM ( SELECT u.Username AS principal FROM `User` AS u INNER JOIN Profile AS p ON p.Id = u.ProfileId WHERE u.IsActive = 1 AND u.IsDeleted = 0 AND p.Name IN ({profile_filter}) UNION SELECT u.Email AS principal FROM `User` AS u INNER JOIN Profile AS p ON p.Id = u.ProfileId WHERE u.IsActive = 1 AND u.IsDeleted = 0 AND p.Name IN ({profile_filter}) ) AS protected_users WHERE principal IS NOT NULL AND TRIM(principal) <> '' ORDER BY normalized_principal """ principals = {line.strip().lower() for line in mysql_query(query).splitlines() if line.strip()} service_username = singleton["serviceUsername"].strip().lower() non_service = principals - {service_username} if not non_service: raise IdentitySyncPolicySeedError( "SCIM policy resolved no active protected administrator principals" ) principals.add(service_username) return sorted(principals) def singleton_sql(singleton: dict[str, Any], exempt_principals: list[str]) -> str: values: dict[str, str | int] = { "Id": "identity-sync-policy-default", "Name": singleton["name"], "DefaultProfileName": singleton["defaultProfileName"], "ServiceProfileName": singleton["serviceProfileName"], "ServiceUsername": singleton["serviceUsername"], "ExemptPrincipals": "\n".join(exempt_principals), "BreakerFailureThreshold": singleton["breakerFailureThreshold"], "BreakerCooldownSeconds": singleton["breakerCooldownSeconds"], "TokenTtlHours": singleton["tokenTtlHours"], "TimeZoneSidKey": singleton["timeZoneSidKey"], "LocaleSidKey": singleton["localeSidKey"], "EmailEncodingKey": singleton["emailEncodingKey"], "LanguageLocaleKey": singleton["languageLocaleKey"], } columns = [*values, "CreatedDate", "LastModifiedDate", "SystemModstamp", "IsDeleted"] rendered_values = [sql_literal(values[column]) for column in values] + [ "NOW()", "NOW()", "NOW()", "0", ] mutable = [column for column in values if column not in {"Id", "Name"}] updates = [f"{column} = VALUES({column})" for column in mutable] updates.extend(["IsDeleted = 0", "LastModifiedDate = NOW()", "SystemModstamp = NOW()"]) return ( "INSERT INTO IdentitySyncSetting (" + ", ".join(columns) + ") VALUES (" + ", ".join(rendered_values) + ") ON DUPLICATE KEY UPDATE " + ", ".join(updates) ) def singleton_readback_sql(singleton: dict[str, Any]) -> str: return ( "SELECT DefaultProfileName, ServiceProfileName, ServiceUsername, ExemptPrincipals, " "BreakerFailureThreshold, BreakerCooldownSeconds, TokenTtlHours, TimeZoneSidKey, " "LocaleSidKey, EmailEncodingKey, LanguageLocaleKey, IsDeleted " "FROM IdentitySyncSetting WHERE Name = " + sql_literal(singleton["name"]) + " LIMIT 1" ) def expected_singleton_row( singleton: dict[str, Any], exempt_principals: list[str], ) -> list[str]: return [ singleton["defaultProfileName"], singleton["serviceProfileName"], singleton["serviceUsername"], "\n".join(exempt_principals), str(singleton["breakerFailureThreshold"]), str(singleton["breakerCooldownSeconds"]), str(singleton["tokenTtlHours"]), singleton["timeZoneSidKey"], singleton["localeSidKey"], singleton["emailEncodingKey"], singleton["languageLocaleKey"], "0", ] def verify_singleton_result( raw: str, singleton: dict[str, Any], exempt_principals: list[str], ) -> None: row = raw.rstrip("\n").split("\t") expected = expected_singleton_row(singleton, exempt_principals) if row != expected: raise IdentitySyncPolicySeedError("SCIM policy readback did not match the L2 Seed") def verify_singleton(singleton: dict[str, Any], exempt_principals: list[str]) -> None: verify_singleton_result( mysql_query(singleton_readback_sql(singleton)), singleton, exempt_principals, ) def preflight_singleton_write( singleton: dict[str, Any], exempt_principals: list[str], ) -> None: # Exercise the exact write/readback inside one rolled-back DB transaction # before creating the Profile. Schema, constraint, and readback defects # therefore fail without leaving durable policy or Profile state. result = mysql_query( "START TRANSACTION;\n" + singleton_sql(singleton, exempt_principals) + ";\n" + singleton_readback_sql(singleton) + ";\nROLLBACK" ) verify_singleton_result(result, singleton, exempt_principals) def write_summary( output_dir: Path, *, dry_run: bool, verification_status: str, duration_ms: int, ) -> None: summary = { "schema_version": 1, "format": SUMMARY_FORMAT, "module_id": MODULE_ID, "dry_run": dry_run, "apply_engine": "runtime-api-mysql-cli", "planned_counts": dict(PLANNED_COUNTS), "applied_counts": { name: 0 if dry_run else count for name, count in PLANNED_COUNTS.items() }, "coverage_complete": True, "verification_status": verification_status, "duration_ms": duration_ms, } output_dir.mkdir(parents=True, exist_ok=False) (output_dir / SUMMARY_FILENAME).write_text( json.dumps(summary, indent=2, sort_keys=True) + "\n", encoding="utf-8" ) print(json.dumps(summary, indent=2, sort_keys=True)) def run_adapter(args: argparse.Namespace) -> None: started = time.perf_counter() source_root = Path(args.source_root).expanduser().resolve(strict=True) if args.module_id != MODULE_ID: raise IdentitySyncPolicySeedError("module id is not the SCIM policy Seed module") if args.invocation_id not in { "staging.identity_sync_policy.post_backend_health", "staging.identity_sync_policy.d3_pre_ingress", "production.identity_sync_policy.pre_candidate_start", }: raise IdentitySyncPolicySeedError("invocation id is not a reviewed SCIM policy Seed route") try: selection = json.loads(args.selection_json) except json.JSONDecodeError as exc: raise IdentitySyncPolicySeedError("selection must be valid JSON") from exc if selection != {}: raise IdentitySyncPolicySeedError("SCIM policy Seed does not accept a selection") seed = load_seed(source_root, args.data_source) singleton = seed["singleton"] onboarding_profile = seed["onboardingProfile"] output_dir = Path(args.output_dir).expanduser().resolve() if args.dry_run: write_summary( output_dir, dry_run=True, verification_status="planned", duration_ms=round((time.perf_counter() - started) * 1000), ) return require_profiles(singleton, include_onboarding=False, include_service=False) exempt_principals = protected_principals(singleton) preflight_singleton_write(singleton, exempt_principals) access_token = runtime_admin_access_token() ensure_onboarding_profile(access_token, singleton, onboarding_profile) service_profile_id = ensure_service_profile( access_token, singleton, seed["serviceProfile"], ) ensure_service_user( access_token, singleton, seed["serviceUser"], service_profile_id, ) require_profiles(singleton) mysql_query(singleton_sql(singleton, exempt_principals)) verify_singleton(singleton, exempt_principals) write_summary( output_dir, dry_run=False, verification_status="passed", duration_ms=round((time.perf_counter() - started) * 1000), ) def main(argv: list[str] | None = None) -> int: try: run_adapter(parse_args(argv)) except (IdentitySyncPolicySeedError, OSError, ValueError) as exc: print(f"SCIM policy Seed adapter failed: {exc}", file=sys.stderr) return 2 return 0 if __name__ == "__main__": raise SystemExit(main())