xinmi/scripts/weknora_historical_identity_etl.py

"""Preserve deleted owner 2 without reconstructing or activating an account.

The caller supplies a consistent read-only SQLite snapshot and executes any
returned SQL in its own transaction. No source credentials are read. This
separate batch does not broaden the enabled-user identity migration.
"""
from __future__ import annotations

import hashlib
import json
import re
import sqlite3

from weknora_identity_etl import (IDENTITY_LEDGER, ORGANIZATION_LEDGER, PREFERENCE_KEY,
                                 SOURCE, SOURCE_ORGANIZATION, TARGET_DATABASE,
                                 TARGET_TENANT, _stable_id, identity_plan)

SOURCE_USER_ID = 2
BATCH = "deleted-owner-2"
TARGET_USERNAME = "\u5df2\u5220\u9664\u7684\u65e7\u7528\u6237 #2"
TARGET_EMAIL = f"legacy-{SOURCE}-t{TARGET_TENANT}-o1-u2@migration.invalid"
OWNER_REFERENCES = {"canvas": [1, 2], "trainingmap": [2, 3]}
POLICY = "historical_owner_only; no_login; no_inferred_identity_permissions_or_grants"


def _source_tombstone() -> dict:
    return {"id": SOURCE_USER_ID, "organization_id": SOURCE_ORGANIZATION,
            "tombstone": True, "source_record_missing": True,
            "owner_references": {key: list(value) for key, value in OWNER_REFERENCES.items()}}


def _historical_user(organization: dict) -> dict:
    source = _source_tombstone()
    provenance = {"schema_version": 1, "source": SOURCE, "batch": BATCH,
                  "target_tenant_id": TARGET_TENANT, "source_organization": organization,
                  "source_user": source, "email_is_placeholder": True,
                  "account_policy": POLICY, "pending_activation": [],
                  "permission_mapping": "none_historical_owner_only"}
    return {"source_user": source, "target_id": _stable_id(SOURCE_USER_ID, TARGET_TENANT),
            "action": "create_suspended", "target_username": TARGET_USERNAME,
            "target_email": TARGET_EMAIL, "email_is_placeholder": True,
            "target_membership_role": "contributor", "is_active": False,
            "membership_status": "suspended", "is_system_admin": False,
            "can_access_all_tenants": False, "pending_activation": [],
            "preferences": {PREFERENCE_KEY: provenance}}


def historical_identity_plan(db: sqlite3.Connection, tenant_id: int) -> dict:
    """Plan only the reviewed, missing owner referenced by four source rows."""
    # Reuse organization/role validation without changing the original batch.
    enabled = identity_plan(db, tenant_id)
    if db.execute("SELECT id FROM appuser WHERE id = ?", (SOURCE_USER_ID,)).fetchone() is not None:
        raise ValueError("Historical owner source record is present; separate identity review required")
    for table, expected_ids in OWNER_REFERENCES.items():
        rows = db.execute(f"SELECT id, organization_id FROM {table} WHERE owner_user_id = ? ORDER BY id",
                          (SOURCE_USER_ID,)).fetchall()
        if rows != [(source_id, SOURCE_ORGANIZATION) for source_id in expected_ids]:
            raise ValueError(f"Historical owner references differ from the reviewed {table} batch")
    organization = enabled["source_organization"]
    plan = {"schema_version": 1, "source": SOURCE, "batch": BATCH,
            "target_database": TARGET_DATABASE, "target_tenant_id": tenant_id,
            "source_organization": organization,
            "source_role_definitions": enabled["source_role_definitions"],
            "users": [_historical_user(organization)], "skipped": [],
            "account_policy": POLICY,
            "password_policy": "no_source_password_read_or_copied; random_non_bcrypt_placeholder_in_sql",
            "repeat_policy": "durable_ledger; no_updates; no_recreation; reject_activation_or_tokens"}
    _validate_plan(plan)
    return plan


def _validate_plan(plan: dict) -> None:
    if (plan.get("schema_version") != 1 or plan.get("source") != SOURCE or plan.get("batch") != BATCH
            or plan.get("target_database") != TARGET_DATABASE
            or type(plan.get("target_tenant_id")) is not int or plan["target_tenant_id"] != TARGET_TENANT
            or plan.get("account_policy") != POLICY):
        raise ValueError("Historical identity plan has an unsupported source or target scope")
    organization = plan.get("source_organization", {})
    if organization.get("id") != SOURCE_ORGANIZATION or organization.get("enabled") != 1:
        raise ValueError("Historical identity plan requires the reviewed source organization")
    definitions = plan.get("source_role_definitions", [])
    roles = {definition.get("role") for definition in definitions}
    if roles != {"admin", "editor", "reviewer", "viewer"} or len(definitions) != len(roles):
        raise ValueError("Historical identity plan must retain the reviewed organization role definitions")
    for definition in definitions:
        permissions = definition.get("permissions")
        if (definition.get("organization_id") != SOURCE_ORGANIZATION
                or not isinstance(permissions, list) or not all(isinstance(p, str) for p in permissions)):
            raise ValueError("Historical identity plan has invalid organization role definitions")
    # Compare canonical JSON so bool/int coercion cannot validate tampered flags.
    if _canonical(plan.get("users")) != _canonical([_historical_user(organization)]):
        raise ValueError("Historical identity plan changes the pinned tombstone or suspended account policy")


def _canonical(value) -> str:
    return json.dumps(value, ensure_ascii=True, sort_keys=True, separators=(",", ":"))


def _payload(plan: dict) -> tuple[str, str]:
    _validate_plan(plan)
    if not re.fullmatch(r"[a-f0-9]{64}", str(plan.get("source_snapshot_sha256", ""))):
        raise ValueError("Historical identity SQL requires the caller's source_snapshot_sha256")
    payload = _canonical(plan)
    return "'" + payload.replace("'", "''") + "'::jsonb", hashlib.sha256(payload.encode()).hexdigest()[:16]


def historical_identity_sql(plan: dict) -> str:
    """Emit insert-only SQL; existing durable ledgers are mandatory."""
    literal, digest = _payload(plan)
    return f"""SET LOCAL standard_conforming_strings = on;
DO $historical_identity_{digest}$
DECLARE
 migration jsonb := {literal};
 item jsonb := migration->'users'->0;
 mapped_user record;
 existing_user users%ROWTYPE;
 existing_member tenant_members%ROWTYPE;
 has_mapping boolean;
 has_user boolean;
BEGIN
 IF current_database() <> '{TARGET_DATABASE}' THEN
  RAISE EXCEPTION 'Historical identity target database mismatch'; END IF;
 PERFORM pg_advisory_xact_lock({TARGET_TENANT}, 11901);
 PERFORM 1 FROM tenants WHERE id={TARGET_TENANT} AND status='active' AND deleted_at IS NULL FOR SHARE;
 IF NOT FOUND THEN RAISE EXCEPTION 'Historical identity requires active target tenant'; END IF;
 LOCK TABLE users, tenant_members, auth_tokens IN SHARE ROW EXCLUSIVE MODE;
 LOCK TABLE {ORGANIZATION_LEDGER}, {IDENTITY_LEDGER} IN SHARE ROW EXCLUSIVE MODE;
 PERFORM 1 FROM {ORGANIZATION_LEDGER}
  WHERE source_system='{SOURCE}' AND source_organization_id={SOURCE_ORGANIZATION}
   AND target_tenant_id={TARGET_TENANT} AND schema_version=1
   AND source_snapshot=migration->'source_organization'
   AND source_role_definitions=migration->'source_role_definitions';
 IF NOT FOUND THEN RAISE EXCEPTION 'Historical identity organization mapping conflict'; END IF;
 SELECT * INTO mapped_user FROM {IDENTITY_LEDGER}
  WHERE source_system='{SOURCE}' AND source_organization_id={SOURCE_ORGANIZATION}
   AND source_user_id={SOURCE_USER_ID};
 has_mapping := FOUND;
 IF has_mapping THEN
  IF mapped_user.target_tenant_id IS DISTINCT FROM {TARGET_TENANT}
   OR mapped_user.target_user_id IS DISTINCT FROM item->>'target_id'
   OR mapped_user.schema_version IS DISTINCT FROM 1
   OR mapped_user.migration_action IS DISTINCT FROM 'create_suspended'
   OR mapped_user.source_snapshot IS DISTINCT FROM item->'source_user'
   OR mapped_user.initial_requested_username IS DISTINCT FROM item->>'target_username'
   OR mapped_user.initial_requested_email IS DISTINCT FROM item->>'target_email'
  THEN RAISE EXCEPTION 'Historical durable identity mapping conflict'; END IF;
 ELSE
  IF EXISTS (SELECT 1 FROM {IDENTITY_LEDGER} WHERE target_tenant_id={TARGET_TENANT}
             AND target_user_id=item->>'target_id')
  THEN RAISE EXCEPTION 'Historical target identity is mapped to another source'; END IF;
  IF EXISTS (SELECT 1 FROM users WHERE id <> item->>'target_id' AND
    (lower(btrim(username))=lower(btrim(item->>'target_username'))
     OR lower(btrim(email))=lower(btrim(item->>'target_email'))))
  THEN RAISE EXCEPTION 'Historical identity username/email collision'; END IF;
 END IF;
 SELECT * INTO existing_user FROM users WHERE id=item->>'target_id' FOR UPDATE;
 has_user := FOUND;
 IF has_user THEN
  IF existing_user.tenant_id IS DISTINCT FROM {TARGET_TENANT}
   OR existing_user.is_system_admin IS DISTINCT FROM false
   OR existing_user.can_access_all_tenants IS DISTINCT FROM false
   OR existing_user.deleted_at IS NOT NULL
  THEN RAISE EXCEPTION 'Historical identity ID/provenance collision'; END IF;
  IF existing_user.is_active IS DISTINCT FROM false
   OR existing_user.password_hash NOT LIKE '!migration-login-disabled:%'
  THEN RAISE EXCEPTION 'Historical identity must remain login-disabled'; END IF;
  IF NOT has_mapping AND (
   existing_user.username IS DISTINCT FROM item->>'target_username'
   OR existing_user.email IS DISTINCT FROM item->>'target_email'
   OR existing_user.preferences->'{PREFERENCE_KEY}' IS DISTINCT FROM item->'preferences'->'{PREFERENCE_KEY}')
  THEN RAISE EXCEPTION 'Unmapped historical identity lacks exact migration provenance'; END IF;
  SELECT * INTO existing_member FROM tenant_members
   WHERE user_id=existing_user.id AND tenant_id={TARGET_TENANT} AND deleted_at IS NULL;
  IF NOT FOUND OR existing_member.role IS DISTINCT FROM 'contributor'
   OR existing_member.status IS DISTINCT FROM 'suspended'
  THEN RAISE EXCEPTION 'Historical identity membership must remain suspended'; END IF;
  IF EXISTS (SELECT 1 FROM tenant_members WHERE user_id=existing_user.id
             AND tenant_id <> {TARGET_TENANT} AND deleted_at IS NULL)
  THEN RAISE EXCEPTION 'Historical identity has unexpected cross-tenant membership'; END IF;
 ELSE
  IF has_mapping THEN
   RAISE EXCEPTION 'Mapped historical user is missing; refusing to recreate a deleted account'; END IF;
  IF EXISTS (SELECT 1 FROM tenant_members WHERE user_id=item->>'target_id')
  THEN RAISE EXCEPTION 'Orphan or deleted membership conflicts with historical identity'; END IF;
  INSERT INTO users (id,username,email,password_hash,tenant_id,is_active,is_system_admin,
                     can_access_all_tenants,preferences,created_at,updated_at)
  VALUES(item->>'target_id',item->>'target_username',item->>'target_email',
   '!migration-login-disabled:' || gen_random_uuid()::text,{TARGET_TENANT},false,false,false,
   item->'preferences',CURRENT_TIMESTAMP,CURRENT_TIMESTAMP);
  INSERT INTO tenant_members(user_id,tenant_id,role,status,joined_at,created_at,updated_at)
  VALUES(item->>'target_id',{TARGET_TENANT},'contributor','suspended',
   CURRENT_TIMESTAMP,CURRENT_TIMESTAMP,CURRENT_TIMESTAMP);
 END IF;
 IF EXISTS (SELECT 1 FROM auth_tokens WHERE user_id=item->>'target_id')
 THEN RAISE EXCEPTION 'Historical identity must not have authentication tokens'; END IF;
 IF NOT has_mapping THEN
  INSERT INTO {IDENTITY_LEDGER}(source_system,source_organization_id,source_user_id,target_tenant_id,
   target_user_id,migration_action,source_snapshot,source_snapshot_sha256,
   initial_requested_username,initial_requested_email)
  VALUES('{SOURCE}',{SOURCE_ORGANIZATION},{SOURCE_USER_ID},{TARGET_TENANT},item->>'target_id',
   'create_suspended',item->'source_user',migration->>'source_snapshot_sha256',
   item->>'target_username',item->>'target_email');
 END IF;
END $historical_identity_{digest}$;
"""


def historical_identity_verification_sql(plan: dict) -> str:
    """Check provenance and disabled state without returning credentials."""
    literal, _ = _payload(plan)
    return f"""WITH migration AS (SELECT {literal} AS body),
planned AS (SELECT body, body->'users'->0 AS item FROM migration)
SELECT {SOURCE_USER_ID} AS source_user_id, item->>'target_id' AS expected_target_user_id,
 current_database()='{TARGET_DATABASE}' AS database_matches,
 coalesce(t.status='active' AND t.deleted_at IS NULL,false) AS tenant_matches,
 coalesce(o.target_tenant_id={TARGET_TENANT} AND o.schema_version=1
  AND o.source_snapshot=body->'source_organization'
  AND o.source_role_definitions=body->'source_role_definitions',false) AS organization_mapping_matches,
 coalesce(l.target_tenant_id={TARGET_TENANT} AND l.target_user_id=item->>'target_id'
  AND l.schema_version=1 AND l.migration_action='create_suspended'
  AND l.source_snapshot=item->'source_user'
  AND l.initial_requested_username=item->>'target_username'
  AND l.initial_requested_email=item->>'target_email',false) AS identity_mapping_matches,
 coalesce(u.id IS NOT NULL AND u.deleted_at IS NULL AND u.tenant_id={TARGET_TENANT}
  AND NOT u.is_system_admin AND NOT u.can_access_all_tenants
  AND NOT EXISTS (SELECT 1 FROM tenant_members other_member WHERE other_member.user_id=u.id
   AND other_member.tenant_id <> {TARGET_TENANT} AND other_member.deleted_at IS NULL),false) AS account_scope_matches,
 coalesce(NOT u.is_active AND u.password_hash LIKE '!migration-login-disabled:%',false) AS login_disabled,
 coalesce(m.role='contributor' AND m.status='suspended',false) AS membership_matches,
 NOT EXISTS (SELECT 1 FROM auth_tokens a WHERE a.user_id=item->>'target_id') AS no_auth_tokens,
 u.username,u.is_active,m.status AS membership_status,
 l.source_snapshot_sha256 AS original_source_snapshot_sha256
FROM planned
LEFT JOIN tenants t ON t.id={TARGET_TENANT}
LEFT JOIN {ORGANIZATION_LEDGER} o ON o.source_system='{SOURCE}' AND o.source_organization_id={SOURCE_ORGANIZATION}
LEFT JOIN {IDENTITY_LEDGER} l ON l.source_system='{SOURCE}' AND l.source_organization_id={SOURCE_ORGANIZATION}
 AND l.source_user_id={SOURCE_USER_ID}
LEFT JOIN users u ON u.id=item->>'target_id'
LEFT JOIN tenant_members m ON m.user_id=u.id AND m.tenant_id={TARGET_TENANT} AND m.deleted_at IS NULL;
"""