Synapse Migration

Security & Governance:
│
│  ── API-Automatable ──────────────────────────────────
├── S1: Identity & Authentication                    (API)
├── S2: Workspace RBAC Mapping (Synapse → Fabric)    (API)
├── S3: Secret Management (Key Vault integration)    (API)
├── S4: Governance & Compliance (Purview, labels)     (API)
│
│  ── Partial / Manual ─────────────────────────────────
├── S5: Data-Level Security (OneLake RBAC, RLS, CLS) (Partial — T-SQL + Portal)
├── S6: Network Security (MPE, Private Link)          (Manual — Portal / Admin)
├── S7: On-Premises Data Gateway (OPDG)               (Manual — Install)
│
│  ── Checklist ────────────────────────────────────────
└── S8: Production Security Checklist

POST https://api.fabric.microsoft.com/v1/workspaces/{workspaceId}/assignWorkspaceIdentity

import requests

fabric_token = get_token("https://api.fabric.microsoft.com")

resp = requests.post(
    f"https://api.fabric.microsoft.com/v1/workspaces/{fabric_ws_id}/assignWorkspaceIdentity",
    headers={"Authorization": f"Bearer {fabric_token}"}
)

if resp.status_code in (200, 201):
    print("Workspace Identity enabled")
elif resp.status_code == 409:
    print("Workspace Identity already assigned")
else:
    print(f"Failed: {resp.status_code} — {resp.text}")

synapse_dev_token = get_token("https://dev.azuresynapse.net")

# List Linked Services via Synapse data-plane API
resp = requests.get(
    f"https://{ws_name}.dev.azuresynapse.net/linkedservices"
    f"?api-version=2020-12-01",
    headers={"Authorization": f"Bearer {synapse_dev_token}"}
)

# Extract ADLS Gen2 storage account names
adls_accounts = set()
for ls in resp.json().get("value", []):
    props = ls.get("properties", {}).get("typeProperties", {})
    url = props.get("url", "") or props.get("endpoint", "")
    # Match https://<account>.dfs.core.windows.net or .blob.core.windows.net
    if ".dfs.core.windows.net" in url or ".blob.core.windows.net" in url:
        account_name = url.split("//")[1].split(".")[0]
        adls_accounts.add(account_name)

print(f"Found {len(adls_accounts)} ADLS accounts referenced by Synapse:")
for acct in sorted(adls_accounts):
    print(f"  - {acct}")

# List all Lakehouses in the target workspace
resp = requests.get(
    f"https://api.fabric.microsoft.com/v1/workspaces/{fabric_ws_id}/items?type=Lakehouse",
    headers={"Authorization": f"Bearer {fabric_token}"}
)

shortcut_accounts = set()
for lh in resp.json().get("value", []):
    lh_id = lh["id"]
    # List shortcuts for this Lakehouse
    sc_resp = requests.get(
        f"https://api.fabric.microsoft.com/v1/workspaces/{fabric_ws_id}"
        f"/items/{lh_id}/shortcuts",
        headers={"Authorization": f"Bearer {fabric_token}"}
    )
    if sc_resp.status_code == 200:
        for sc in sc_resp.json().get("value", []):
            target = sc.get("target", {})
            if "adlsGen2" in target:
                loc = target["adlsGen2"].get("location", "")
                if ".dfs.core.windows.net" in loc:
                    acct = loc.split("//")[1].split(".")[0]
                    shortcut_accounts.add(acct)

all_accounts = adls_accounts | shortcut_accounts
print(f"\nAll ADLS accounts needing Storage Blob Data Reader ({len(all_accounts)}):")
for acct in sorted(all_accounts):
    source = []
    if acct in adls_accounts: source.append("Linked Service")
    if acct in shortcut_accounts: source.append("Shortcut")
    print(f"  - {acct}  (from: {', '.join(source)})")

import uuid

# Storage Blob Data Reader role definition ID
STORAGE_BLOB_DATA_READER = "2a2b9908-6ea1-4ae2-8e65-a410df84e7d1"

# Get the Fabric Workspace Identity's service principal ID
ws_resp = requests.get(
    f"https://api.fabric.microsoft.com/v1/workspaces/{fabric_ws_id}",
    headers={"Authorization": f"Bearer {fabric_token}"}
)
ws_identity_id = ws_resp.json().get("identity", {}).get("servicePrincipalId")

if not ws_identity_id:
    print("ERROR: Workspace Identity not found. Run Step 1 first.")
else:
    results = {"success": 0, "exists": 0, "failed": 0}

    for acct in all_accounts:
        storage_scope = (
            f"/subscriptions/{sub_id}/resourceGroups/{rg}"
            f"/providers/Microsoft.Storage/storageAccounts/{acct}"
        )
        assignment_id = str(uuid.uuid4())

        resp = requests.put(
            f"https://management.azure.com{storage_scope}"
            f"/providers/Microsoft.Authorization/roleAssignments/{assignment_id}"
            f"?api-version=2022-04-01",
            headers={
                "Authorization": f"Bearer {arm_token}",
                "Content-Type": "application/json"
            },
            json={
                "properties": {
                    "roleDefinitionId": f"/subscriptions/{sub_id}/providers/Microsoft.Authorization/roleDefinitions/{STORAGE_BLOB_DATA_READER}",
                    "principalId": ws_identity_id,
                    "principalType": "ServicePrincipal"
                }
            }
        )

        if resp.status_code in (200, 201):
            results["success"] += 1
            print(f"  ✓ Granted on {acct}")
        elif resp.status_code == 409:
            results["exists"] += 1
            print(f"  ✓ Already granted on {acct}")
        else:
            results["failed"] += 1
            print(f"  ✗ Failed on {acct}: {resp.status_code} — {resp.text}")

    print(f"\nResults: {results['success']} granted, {results['exists']} already existed, {results['failed']} failed")

import re

SQL_AUTH_PATTERNS = [
    (r"jdbc:.*User=|jdbc:.*Password=", "JDBC with SQL credentials"),
    (r"\.option\([\"']user[\"']", "Spark JDBC .option('user')"),
    (r"\.option\([\"']password[\"']", "Spark JDBC .option('password')"),
    (r"SqlConnection\(.*Uid=|SqlConnection\(.*Pwd=", ".NET SqlConnection with SQL auth"),
    (r"pyodbc\.connect\(.*UID=|pyodbc\.connect\(.*PWD=", "pyodbc with SQL auth"),
]

# notebooks = list of exported notebook dicts (from Phase 2 inventory)
sql_auth_issues = []

for nb in notebooks:
    nb_name = nb.get("name", "unknown")
    cells = nb.get("properties", {}).get("cells", [])
    for i, cell in enumerate(cells):
        source = "".join(cell.get("source", []))
        for pattern, description in SQL_AUTH_PATTERNS:
            if re.search(pattern, source, re.IGNORECASE):
                sql_auth_issues.append({
                    "notebook": nb_name,
                    "cell": i + 1,
                    "issue": description
                })

if sql_auth_issues:
    print(f"⚠ Found {len(sql_auth_issues)} SQL auth usage(s) — must convert to Entra:")
    for issue in sql_auth_issues:
        print(f"  {issue['notebook']} (cell {issue['cell']}): {issue['issue']}")
    print("\n  → Replace with Entra token: notebookutils.credentials.getToken('https://database.windows.net')")
else:
    print("✓ No SQL authentication patterns found in notebooks")

GET https://api.fabric.microsoft.com/v1/workspaces/{workspaceId}/roleAssignments

POST https://api.fabric.microsoft.com/v1/workspaces/{workspaceId}/roleAssignments

{
  "principal": {
    "id": "{entraObjectId}",
    "type": "User"
  },
  "role": "Contributor"
}

PATCH https://api.fabric.microsoft.com/v1/workspaces/{workspaceId}/roleAssignments/{roleAssignmentId}

{
  "role": "Member"
}

DELETE https://api.fabric.microsoft.com/v1/workspaces/{workspaceId}/roleAssignments/{roleAssignmentId}

import requests

arm_token = get_token("https://management.azure.com")
synapse_scope = (
    f"/subscriptions/{sub_id}/resourceGroups/{rg}"
    f"/providers/Microsoft.Synapse/workspaces/{ws_name}"
)

# List all role assignments on the Synapse workspace
resp = requests.get(
    f"https://management.azure.com{synapse_scope}"
    f"/providers/Microsoft.Authorization/roleAssignments"
    f"?api-version=2022-04-01&$filter=atScope()",
    headers={"Authorization": f"Bearer {arm_token}"}
)
synapse_assignments = resp.json()["value"]

# Azure RBAC role GUIDs scoped to Synapse workspaces
# Note: Synapse Studio RBAC roles (Apache Spark Administrator, Synapse SQL Administrator,
# Linked Data Manager, Credential User) are managed via the Synapse RBAC API at
# https://{ws}.dev.azuresynapse.net/rbac/roleAssignments — they do NOT appear in ARM
# roleAssignments. The mapping below covers Azure RBAC roles only.
SYNAPSE_ROLES = {
    "6e4bf58a-b8e1-4cc3-bbf9-d73143322b78": "Synapse Administrator",
    "7af0c69a-a548-47d6-aea3-d00e69bd83aa": "Synapse SQL Administrator",
    "c3a6d2f1-a26f-4810-9b0f-591308d5cbf1": "Synapse Contributor",
    "b5c23728-2989-4e2a-9958-2c22a2689f09": "Synapse Monitoring Operator",
    "2a5c394f-5eb7-4d4f-9c8e-e8eae39faebc": "Synapse User",
}

def resolve_synapse_role(role_def_id):
    """Extract the role GUID and map to Synapse role name."""
    role_guid = role_def_id.split("/")[-1]
    return SYNAPSE_ROLES.get(role_guid, f"Unknown ({role_guid})")

# Azure RBAC Synapse role → Fabric role mapping
ROLE_MAP = {
    "Synapse Administrator":       "Admin",
    "Synapse SQL Administrator":   "Admin",
    "Synapse Contributor":         "Contributor",
    "Synapse Monitoring Operator": "Viewer",
    "Synapse User":                "Viewer",
}
# Note: Synapse Studio RBAC roles (Apache Spark Administrator, Linked Data Manager,
# Credential User) are not exported via ARM. If needed, query the Synapse RBAC API
# at https://{ws}.dev.azuresynapse.net/rbac/roleAssignments separately.

# Build the migration plan
migration_plan = []
skipped = []

for assignment in synapse_assignments:
    props = assignment["properties"]
    synapse_role = resolve_synapse_role(props["roleDefinitionId"])
    fabric_role = ROLE_MAP.get(synapse_role)

    entry = {
        "principalId":   props["principalId"],
        "principalType": props.get("principalType", "User"),
        "synapseRole":   synapse_role,
        "fabricRole":    fabric_role,
    }

    if fabric_role is None:
        skipped.append(entry)
    else:
        migration_plan.append(entry)

print("RBAC Migration Plan:")
print(f"  Assignments to migrate: {len(migration_plan)}")
print(f"  Assignments skipped:    {len(skipped)}")
print()
print(f"  {'Principal ID':<40} {'Synapse Role':<30} {'Fabric Role':<15}")
print(f"  {'─'*40} {'─'*30} {'─'*15}")
for entry in migration_plan:
    print(f"  {entry['principalId']:<40} {entry['synapseRole']:<30} {entry['fabricRole']:<15}")

if skipped:
    print()
    print("  Skipped (no Fabric equivalent):")
    for entry in skipped:
        print(f"    {entry['principalId']} — {entry['synapseRole']}")
        print(f"      → Action: Configure Key Vault RBAC instead (see S3)")

fabric_token = get_token("https://api.fabric.microsoft.com")

# Deduplicate — if a principal has multiple Synapse roles, take the highest Fabric role
ROLE_PRIORITY = {"Admin": 4, "Member": 3, "Contributor": 2, "Viewer": 1}

principal_roles = {}
for entry in migration_plan:
    pid = entry["principalId"]
    role = entry["fabricRole"]
    if pid not in principal_roles or ROLE_PRIORITY[role] > ROLE_PRIORITY[principal_roles[pid]["fabricRole"]]:
        principal_roles[pid] = entry

# Create assignments
results = {"success": 0, "failed": 0, "errors": []}

for pid, entry in principal_roles.items():
    resp = requests.post(
        f"https://api.fabric.microsoft.com/v1/workspaces/{fabric_ws_id}/roleAssignments",
        headers={
            "Authorization": f"Bearer {fabric_token}",
            "Content-Type": "application/json"
        },
        json={
            "principal": {
                "id": entry["principalId"],
                "type": entry["principalType"]
            },
            "role": entry["fabricRole"]
        }
    )

    if resp.status_code in (200, 201):
        results["success"] += 1
    elif resp.status_code == 409:
        # Assignment already exists — update instead
        # First, list existing to get the roleAssignmentId
        existing = requests.get(
            f"https://api.fabric.microsoft.com/v1/workspaces/{fabric_ws_id}/roleAssignments",
            headers={"Authorization": f"Bearer {fabric_token}"}
        ).json()["value"]
        match = next((a for a in existing if a["principal"]["id"] == pid), None)
        if match:
            requests.patch(
                f"https://api.fabric.microsoft.com/v1/workspaces/{fabric_ws_id}/roleAssignments/{match['id']}",
                headers={
                    "Authorization": f"Bearer {fabric_token}",
                    "Content-Type": "application/json"
                },
                json={"role": entry["fabricRole"]}
            )
            results["success"] += 1
    else:
        results["failed"] += 1
        results["errors"].append({
            "principal": pid, "role": entry["fabricRole"],
            "status": resp.status_code, "error": resp.text
        })

print(f"\nRBAC Migration Results: {results['success']} succeeded, {results['failed']} failed")
for err in results["errors"]:
    print(f"  FAILED: {err['principal']} → {err['role']} — {err['status']}: {err['error']}")

# List all Fabric role assignments and compare
resp = requests.get(
    f"https://api.fabric.microsoft.com/v1/workspaces/{fabric_ws_id}/roleAssignments",
    headers={"Authorization": f"Bearer {fabric_token}"}
)
fabric_assignments = resp.json()["value"]

print(f"\nFabric Workspace Role Assignments ({len(fabric_assignments)} total):")
print(f"  {'Display Name':<30} {'Type':<20} {'Role':<15}")
print(f"  {'─'*30} {'─'*20} {'─'*15}")
for a in fabric_assignments:
    print(f"  {a['principal'].get('displayName', 'N/A'):<30} {a['principal']['type']:<20} {a['role']:<15}")

PUT https://management.azure.com/subscriptions/{subId}/resourceGroups/{rg}/providers/Microsoft.KeyVault/vaults/{vaultName}/providers/Microsoft.Authorization/roleAssignments/{assignmentId}?api-version=2022-04-01

{
  "properties": {
    "roleDefinitionId": "/subscriptions/{subId}/providers/Microsoft.Authorization/roleDefinitions/4633458b-17de-408a-b874-0445c86b69e6",
    "principalId": "{workspaceIdentityObjectId}",
    "principalType": "ServicePrincipal"
  }
}

import re

# notebooks = list of exported notebook dicts (from Phase 2 inventory)
vault_secrets = {}  # vault_url → set of secret names
linked_services = set()  # Linked Service names (need manual resolution)

for nb in notebooks:
    nb_name = nb.get("name", "unknown")
    cells = nb.get("properties", {}).get("cells", [])
    for i, cell in enumerate(cells):
        source = "".join(cell.get("source", []))

        # Match getSecret calls with vault URL + secret name
        for match in re.finditer(
            r'getSecret\s*\(\s*["\']([^"\']+vault\.azure\.net[^"\']*)["\'],?\s*["\']([^"\']*)["\']',
            source
        ):
            vault_url = match.group(1).rstrip("/")
            secret_name = match.group(2)
            vault_secrets.setdefault(vault_url, set()).add(secret_name)
            print(f"  {nb_name} (cell {i+1}): getSecret → {vault_url} / {secret_name}")

        # Match TokenLibrary.getSecret
        for match in re.finditer(
            r'TokenLibrary\.getSecret\s*\(\s*["\']([^"\']+vault\.azure\.net[^"\']*)["\'],?\s*["\']([^"\']*)["\']',
            source
        ):
            vault_url = match.group(1).rstrip("/")
            secret_name = match.group(2)
            vault_secrets.setdefault(vault_url, set()).add(secret_name)
            print(f"  {nb_name} (cell {i+1}): TokenLibrary.getSecret → {vault_url} / {secret_name}")

        # Match getConnectionStringOrCreds (Linked Service — needs manual vault mapping)
        for match in re.finditer(
            r'getConnectionStringOrCreds\s*\(\s*["\']([^"\']+)["\']',
            source
        ):
            ls_name = match.group(1)
            linked_services.add(ls_name)
            print(f"  {nb_name} (cell {i+1}): getConnectionStringOrCreds → LinkedService: {ls_name}")

print(f"\nSummary:")
print(f"  Key Vaults discovered: {len(vault_secrets)}")
for vault_url, secrets in vault_secrets.items():
    print(f"    {vault_url}: {len(secrets)} secrets ({', '.join(sorted(secrets))})")

if linked_services:
    print(f"\n  ⚠ Linked Services still using getConnectionStringOrCreds ({len(linked_services)}):")
    for ls in sorted(linked_services):
        print(f"    - {ls}")
    print("    → These must be migrated to getSecret() first (see connector-refactoring.md)")

vault_resources = {}

for vault_url in vault_secrets:
    # Extract vault name from URL: https://myvault.vault.azure.net → myvault
    vault_name = vault_url.replace("https://", "").split(".")[0]

    # Resolve the vault's resource ID via ARM (try known resource group first, then search)
    resp = requests.get(
        f"https://management.azure.com/subscriptions/{sub_id}/resourceGroups/{rg}"
        f"/providers/Microsoft.KeyVault/vaults/{vault_name}?api-version=2022-07-01",
        headers={"Authorization": f"Bearer {arm_token}"}
    )

    if resp.status_code == 200:
        vault_resources[vault_name] = resp.json()["id"]
    else:
        # Search across all resource groups
        search_resp = requests.get(
            f"https://management.azure.com/subscriptions/{sub_id}"
            f"/resources?$filter=resourceType eq 'Microsoft.KeyVault/vaults'"
            f" and name eq '{vault_name}'&api-version=2021-04-01",
            headers={"Authorization": f"Bearer {arm_token}"}
        )
        matches = search_resp.json().get("value", [])
        if matches:
            vault_resources[vault_name] = matches[0]["id"]
        else:
            print(f"  ⚠ Could not find vault: {vault_name}")

print(f"\nResolved {len(vault_resources)} vault resource IDs")

import uuid

KEY_VAULT_SECRETS_USER = "4633458b-17de-408a-b874-0445c86b69e6"

# Get Workspace Identity service principal ID (from S1 Step 4)
ws_resp = requests.get(
    f"https://api.fabric.microsoft.com/v1/workspaces/{fabric_ws_id}",
    headers={"Authorization": f"Bearer {fabric_token}"}
)
ws_identity_id = ws_resp.json().get("identity", {}).get("servicePrincipalId")

if not ws_identity_id:
    print("ERROR: Workspace Identity not enabled. Run S1 first.")
else:
    results = {"granted": 0, "exists": 0, "failed": 0}

    for vault_name, vault_resource_id in vault_resources.items():
        assignment_id = str(uuid.uuid4())

        resp = requests.put(
            f"https://management.azure.com{vault_resource_id}"
            f"/providers/Microsoft.Authorization/roleAssignments/{assignment_id}"
            f"?api-version=2022-04-01",
            headers={
                "Authorization": f"Bearer {arm_token}",
                "Content-Type": "application/json"
            },
            json={
                "properties": {
                    "roleDefinitionId": f"/subscriptions/{sub_id}/providers/Microsoft.Authorization/roleDefinitions/{KEY_VAULT_SECRETS_USER}",
                    "principalId": ws_identity_id,
                    "principalType": "ServicePrincipal"
                }
            }
        )

        if resp.status_code in (200, 201):
            results["granted"] += 1
            print(f"  ✓ Granted Key Vault Secrets User on {vault_name}")
        elif resp.status_code == 409:
            results["exists"] += 1
            print(f"  ✓ Already granted on {vault_name}")
        else:
            results["failed"] += 1
            print(f"  ✗ Failed on {vault_name}: {resp.status_code} — {resp.text}")

    print(f"\nKey Vault access: {results['granted']} granted, {results['exists']} existed, {results['failed']} failed")

# Test secret retrieval for each vault (requires running from a Fabric notebook)
# This step must be executed inside a Fabric notebook after deployment:

# for vault_url, secrets in vault_secrets.items():
#     for secret_name in secrets:
#         try:
#             val = notebookutils.credentials.getSecret(vault_url, secret_name)
#             print(f"  ✓ {vault_url}/{secret_name} — accessible")
#         except Exception as e:
#             print(f"  ✗ {vault_url}/{secret_name} — {e}")

PATCH https://api.fabric.microsoft.com/v1/workspaces/{workspaceId}/items/{itemId}

{
  "sensitivityLabel": {
    "labelId": "{mipLabelId}"
  }
}

POST https://api.fabric.microsoft.com/v1/workspaces/{workspaceId}/items/{itemId}/endorse

{
  "endorsement": "Certified"
}

-- Run against Synapse Dedicated SQL Pool
-- Export RLS security policies, filter functions, and predicate bindings

-- 1a. List all security policies
SELECT
    p.name AS policy_name,
    p.is_enabled,
    SCHEMA_NAME(p.schema_id) AS policy_schema
FROM sys.security_policies p
ORDER BY p.name;

-- 1b. List all filter predicate functions
SELECT
    SCHEMA_NAME(o.schema_id) AS function_schema,
    o.name AS function_name,
    m.definition AS function_definition
FROM sys.objects o
JOIN sys.sql_modules m ON o.object_id = m.object_id
WHERE o.type = 'IF'  -- Inline table-valued function
ORDER BY o.name;

-- 1c. List all security predicates (bindings)
SELECT
    sp.security_policy_id,
    pol.name AS policy_name,
    sp.predicate_type_desc,
    SCHEMA_NAME(t.schema_id) + '.' + t.name AS target_table,
    sp.predicate_definition AS filter_predicate
FROM sys.security_predicates sp
JOIN sys.security_policies pol ON sp.security_policy_id = pol.object_id
JOIN sys.objects t ON sp.target_object_id = t.object_id
ORDER BY pol.name;

-- Run against Synapse Dedicated SQL Pool
-- Export column-level GRANT/DENY permissions

SELECT
    dp.state_desc AS permission_state,  -- GRANT or DENY
    dp.permission_name,
    SCHEMA_NAME(o.schema_id) + '.' + o.name AS table_name,
    c.name AS column_name,
    pr.name AS grantee_name,
    pr.type_desc AS grantee_type
FROM sys.database_permissions dp
JOIN sys.objects o ON dp.major_id = o.object_id
JOIN sys.columns c ON dp.major_id = c.object_id AND dp.minor_id = c.column_id
JOIN sys.database_principals pr ON dp.grantee_principal_id = pr.principal_id
WHERE dp.minor_id > 0  -- Column-level permissions only
ORDER BY table_name, column_name, grantee_name;

-- Run against Synapse Dedicated SQL Pool
-- Export Dynamic Data Masking definitions

SELECT
    SCHEMA_NAME(t.schema_id) + '.' + t.name AS table_name,
    c.name AS column_name,
    c.masking_function AS mask_function
FROM sys.masked_columns c
JOIN sys.tables t ON c.object_id = t.object_id
WHERE c.is_masked = 1
ORDER BY table_name, column_name;

import pyodbc

# Connect to Synapse Dedicated SQL Pool
synapse_conn = pyodbc.connect(
    f"DRIVER={{ODBC Driver 18 for SQL Server}};"
    f"SERVER={synapse_sql_endpoint};"
    f"DATABASE={dedicated_pool_name};"
    f"Authentication=ActiveDirectoryDefault;"
)
cursor = synapse_conn.cursor()

fabric_sql_statements = []

# --- RLS: Export filter functions + security policies ---
cursor.execute("""
    SELECT SCHEMA_NAME(o.schema_id) AS fn_schema, o.name AS fn_name, m.definition
    FROM sys.objects o
    JOIN sys.sql_modules m ON o.object_id = m.object_id
    WHERE o.type = 'IF'
""")
for row in cursor.fetchall():
    fabric_sql_statements.append(f"-- RLS filter function: {row.fn_schema}.{row.fn_name}")
    fabric_sql_statements.append(row.definition + ";")
    fabric_sql_statements.append("")

cursor.execute("""
    SELECT pol.name AS policy_name, pol.is_enabled,
           sp.predicate_type_desc,
           SCHEMA_NAME(t.schema_id) + '.' + t.name AS target_table,
           sp.predicate_definition AS filter_predicate
    FROM sys.security_predicates sp
    JOIN sys.security_policies pol ON sp.security_policy_id = pol.object_id
    JOIN sys.objects t ON sp.target_object_id = t.object_id
    ORDER BY pol.name
""")
policies = {}
for row in cursor.fetchall():
    if row.policy_name not in policies:
        policies[row.policy_name] = {"enabled": row.is_enabled, "predicates": []}
    policies[row.policy_name]["predicates"].append({
        "type": row.predicate_type_desc,
        "table": row.target_table,
        "predicate": row.filter_predicate
    })

for policy_name, policy in policies.items():
    predicates = ", ".join(
        f"ADD {p['type']} PREDICATE {p['predicate']} ON {p['table']}"
        for p in policy["predicates"]
    )
    state = "ON" if policy["enabled"] else "OFF"
    fabric_sql_statements.append(f"-- RLS policy: {policy_name}")
    fabric_sql_statements.append(f"CREATE SECURITY POLICY {policy_name}")
    fabric_sql_statements.append(f"    {predicates}")
    fabric_sql_statements.append(f"    WITH (STATE = {state});")
    fabric_sql_statements.append("")

# --- CLS: Export column-level grants ---
cursor.execute("""
    SELECT dp.state_desc, dp.permission_name,
           SCHEMA_NAME(o.schema_id) + '.' + o.name AS table_name,
           c.name AS column_name, pr.name AS grantee
    FROM sys.database_permissions dp
    JOIN sys.objects o ON dp.major_id = o.object_id
    JOIN sys.columns c ON dp.major_id = c.object_id AND dp.minor_id = c.column_id
    JOIN sys.database_principals pr ON dp.grantee_principal_id = pr.principal_id
    WHERE dp.minor_id > 0
    ORDER BY table_name, column_name
""")
for row in cursor.fetchall():
    fabric_sql_statements.append(
        f"-- CLS: {row.state_desc} {row.permission_name} on {row.table_name}.{row.column_name}")
    fabric_sql_statements.append(
        f"{row.state_desc} {row.permission_name} ON {row.table_name}({row.column_name}) TO [{row.grantee}];")
    fabric_sql_statements.append("")

# --- DDM: Export masking definitions ---
cursor.execute("""
    SELECT SCHEMA_NAME(t.schema_id) + '.' + t.name AS table_name,
           c.name AS column_name, c.masking_function
    FROM sys.masked_columns c
    JOIN sys.tables t ON c.object_id = t.object_id
    WHERE c.is_masked = 1
""")
for row in cursor.fetchall():
    fabric_sql_statements.append(
        f"-- DDM: Mask {row.table_name}.{row.column_name}")
    fabric_sql_statements.append(
        f"ALTER TABLE {row.table_name} ALTER COLUMN {row.column_name} "
        f"ADD MASKED WITH (FUNCTION = '{row.masking_function}');")
    fabric_sql_statements.append("")

synapse_conn.close()

# Write output
output = "\n".join(fabric_sql_statements)
print(f"Generated {len(fabric_sql_statements)} T-SQL statements")
print(output)

# Save to file for review before execution
with open("fabric-security-policies.sql", "w") as f:
    f.write(output)
print("\nSaved to fabric-security-policies.sql — review before executing on Fabric SQL endpoint")

# Review the generated SQL first, then execute against Fabric SQL endpoint
sqlcmd -S {fabric_sql_endpoint} -d {warehouse_or_lakehouse_name} \
    -G -i fabric-security-policies.sql