378 lines
25 KiB
Python
378 lines
25 KiB
Python
|
#!/usr/bin/env python3
|
||
|
|
"""Read-only access migration planner; emitted SQL rolls back unless requested.
|
||
|
|
|
||
|
|
No database writes are executed by this script. Apply the reviewed SQL separately
|
||
|
|
after migrations 000196 and 000207. Repeats use a durable ledger and preserve native edits.
|
||
|
|
"""
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import argparse
|
||
|
|
import hashlib
|
||
|
|
import json
|
||
|
|
import uuid
|
||
|
|
from pathlib import Path
|
||
|
|
|
||
|
|
from apply_weknora_identities import sql_run
|
||
|
|
from weknora_etl import consistent_snapshot
|
||
|
|
from weknora_identity_etl import ADMIN_TARGET_ID, TARGET_DATABASE, TARGET_TENANT
|
||
|
|
|
||
|
|
SOURCE = "yawan-workbench"
|
||
|
|
LEDGER = "yawan_legacy_access_mappings"
|
||
|
|
NAMESPACE = uuid.UUID("93bda6b1-9b55-41a6-ae07-07f844762fab")
|
||
|
|
ACCESS_TABLES = tuple("workbench_access_" + s for s in ("roles", "assignments", "publications", "grants"))
|
||
|
|
GRANT_TABLES = ("knowledgebasegrant", "canvasgrant", "memorygrant", "digitalhumangrant", "agentskillgrant", "agentgrant")
|
||
|
|
EXPANSIONS = {
|
||
|
|
"read": {"read"}, "review": {"tasks:review", "memories:review"},
|
||
|
|
"audit:read": {"logs:read", "audit:read"},
|
||
|
|
"feature:memories": {"feature:agents"}, "feature:skills": {"feature:agents"},
|
||
|
|
"feature:tags": {"feature:knowledge"}, "tags:read": {"kb:read"},
|
||
|
|
"content:write": {"tasks:read", "tasks:write", "tasks:deliver", "kb:read", "kb:write",
|
||
|
|
"agents:read", "agents:write", "digital_humans:read", "flows:read", "memories:read", "memories:write"},
|
||
|
|
"config:read": {"logs:read", "runs:read", "audit:read", "usage:read", "system_logs:read", "ops:read", "ops:manage"},
|
||
|
|
"config:write": {"providers:write", "ops:manage"},
|
||
|
|
"admin:manage": {"roles:manage", "audit:manage", "users:manage", "providers:write", "ops:manage", "billing:manage"},
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
def canonical(value):
|
||
|
|
return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=True, allow_nan=False)
|
||
|
|
|
||
|
|
|
||
|
|
def digest(value):
|
||
|
|
return hashlib.sha256(canonical(value).encode()).hexdigest()
|
||
|
|
|
||
|
|
|
||
|
|
def stable(role):
|
||
|
|
return str(uuid.uuid5(NAMESPACE, f"{SOURCE}:tenant:{TARGET_TENANT}:role:{role}"))
|
||
|
|
|
||
|
|
|
||
|
|
def read_rows(db, table, columns):
|
||
|
|
query = db.execute(f"SELECT {columns} FROM {table} ORDER BY id")
|
||
|
|
names = [column[0] for column in query.description]
|
||
|
|
return [dict(zip(names, row)) for row in query]
|
||
|
|
|
||
|
|
|
||
|
|
def read_source(db):
|
||
|
|
if db.execute("PRAGMA query_only").fetchone()[0] != 1:
|
||
|
|
raise ValueError("A read-only consistent snapshot is required")
|
||
|
|
roles = read_rows(db, "roledefinition", "id,role,label,description,permissions,builtin,organization_id,updated_at")
|
||
|
|
for role in roles:
|
||
|
|
role["permissions"] = json.loads(role["permissions"])
|
||
|
|
resources = {}
|
||
|
|
for table in ("agent", "agentskill", "canvas", "knowledgebase", "digitalhuman", "memory"):
|
||
|
|
extra = ",status" if table == "agent" else ",enabled" if table == "agentskill" else ""
|
||
|
|
resources[table] = read_rows(db, table, "id,name,organization_id,owner_user_id,visibility" + extra) if table != "memory" else []
|
||
|
|
return {"roles": roles, "users": read_rows(db, "appuser", "id,username,role,organization_id,enabled"),
|
||
|
|
"resources": resources, "memory_count": db.execute("SELECT count(*) FROM memory").fetchone()[0],
|
||
|
|
"grants": {table: read_rows(db, table, "*") for table in GRANT_TABLES}}
|
||
|
|
|
||
|
|
|
||
|
|
def role_permissions(source):
|
||
|
|
"""Intersect grouped native actions with actual legacy endpoint checks.
|
||
|
|
|
||
|
|
Content delivery is separate; native Agent management still includes memories.
|
||
|
|
Every old action represented by the grouped permission must be present.
|
||
|
|
"""
|
||
|
|
raw = source["permissions"]
|
||
|
|
if not isinstance(raw, list) or not all(isinstance(p, str) for p in raw):
|
||
|
|
raise ValueError("Malformed legacy role permission list")
|
||
|
|
if "*" in raw:
|
||
|
|
return [], ["administrator_mapping_retained_without_new_custom_role"]
|
||
|
|
permissions = set().union(*(EXPANSIONS.get(p, {p}) for p in raw))
|
||
|
|
result = []
|
||
|
|
requirements = {
|
||
|
|
"content.read": {"tasks:review"},
|
||
|
|
"content.write": {"tasks:write"},
|
||
|
|
"content.review": {"tasks:review"},
|
||
|
|
"content.deliver": {"tasks:deliver"},
|
||
|
|
"flows.read": {"agents:read"}, "flows.write": {"agents:write"},
|
||
|
|
"digital-humans.read": {"read"}, "digital-humans.write": {"agents:write"},
|
||
|
|
"knowledge.read": {"kb:read"}, "knowledge.write": {"kb:write"},
|
||
|
|
"agents.read": {"agents:read", "memories:write"},
|
||
|
|
"agents.write": {"agents:write", "memories:write", "memories:review"},
|
||
|
|
}
|
||
|
|
gates = {
|
||
|
|
"content": {"feature:tasks", "tasks:read", "tasks:write", "tasks:review", "tasks:deliver"},
|
||
|
|
"flows": {"feature:flows", "flows:read", "agents:write"},
|
||
|
|
"digital-humans": {"feature:digital-humans", "digital_humans:read", "agents:write"},
|
||
|
|
"knowledge": {"feature:knowledge", "kb:read"},
|
||
|
|
"agents": {"feature:agents", "agents:read", "agents:write"},
|
||
|
|
}
|
||
|
|
for permission, required in requirements.items():
|
||
|
|
module, action = permission.split(".")
|
||
|
|
if required <= permissions and permissions & gates[module]:
|
||
|
|
if action == "read" or module + ".read" in result:
|
||
|
|
result.append(permission)
|
||
|
|
notes = ["new_modules_chat_sharing_settings_access_not_implicitly_granted"]
|
||
|
|
if permissions & {"agents:read", "memories:review"} and "agents.read" not in result:
|
||
|
|
notes.append("agents_read_withheld_because_native_group_includes_memory_access")
|
||
|
|
return sorted(result), notes
|
||
|
|
|
||
|
|
|
||
|
|
def read_target():
|
||
|
|
# Only authentication-independent metadata is selected, never password/token/model bodies.
|
||
|
|
return json.loads(sql_run(f"""BEGIN ISOLATION LEVEL REPEATABLE READ READ ONLY;
|
||
|
|
SELECT json_build_object(
|
||
|
|
'database',current_database(),
|
||
|
|
'access_tables',(SELECT coalesce(json_agg(tablename ORDER BY tablename),'[]') FROM pg_tables WHERE schemaname='public' AND tablename LIKE 'workbench_access_%'),
|
||
|
|
'content_permissions_ready',to_regclass('workbench_content_permission_migrations') IS NOT NULL,
|
||
|
|
'organization',(SELECT to_jsonb(o) FROM yawan_legacy_organization_mappings o WHERE target_tenant_id={TARGET_TENANT}),
|
||
|
|
'identities',(SELECT json_agg(json_build_object('source_user_id',l.source_user_id,'target_user_id',l.target_user_id,
|
||
|
|
'source_snapshot',l.source_snapshot,'user_tenant',u.tenant_id,'is_active',u.is_active,'is_system_admin',u.is_system_admin,
|
||
|
|
'can_access_all_tenants',u.can_access_all_tenants,'user_deleted',u.deleted_at,
|
||
|
|
'memberships',(SELECT json_agg(json_build_object('tenant_id',m.tenant_id,'role',m.role,'status',m.status,'deleted_at',m.deleted_at)) FROM tenant_members m WHERE m.user_id=u.id)))
|
||
|
|
FROM yawan_legacy_identity_mappings l LEFT JOIN users u ON u.id=l.target_user_id WHERE l.target_tenant_id={TARGET_TENANT}),
|
||
|
|
'asset_mappings',(SELECT json_agg(json_build_object('source_kind',source_kind,'source_id',source_id,'target_id',target_id,
|
||
|
|
'source_visibility',source_snapshot->>'visibility')) FROM yawan_legacy_asset_mappings WHERE target_tenant_id={TARGET_TENANT} AND source_kind IN ('agent','skill')),
|
||
|
|
'flow_mappings',(SELECT json_agg(json_build_object('source_id',source_id,'target_id',target_id,'source_visibility',source_snapshot->>'visibility'))
|
||
|
|
FROM yawan_legacy_business_mappings WHERE target_tenant_id={TARGET_TENANT} AND source_kind='canvas'),
|
||
|
|
'resources',(SELECT json_agg(x) FROM (
|
||
|
|
SELECT 'agents' AS kind,id,tenant_id,created_by AS owner_id,visibility,enabled FROM custom_agents WHERE tenant_id={TARGET_TENANT} AND deleted_at IS NULL
|
||
|
|
UNION ALL SELECT 'skills',id,tenant_id,created_by,visibility,enabled FROM tenant_skill_catalog WHERE tenant_id={TARGET_TENANT} AND instruction_only=true AND deleted_at IS NULL
|
||
|
|
UNION ALL SELECT 'knowledge',id,tenant_id,creator_id,'global',true FROM knowledge_bases WHERE tenant_id={TARGET_TENANT} AND deleted_at IS NULL AND is_temporary=false
|
||
|
|
UNION ALL SELECT 'flows',id,tenant_id,owner_id,'private',true FROM workbench_resources WHERE tenant_id={TARGET_TENANT} AND kind='flows') x));
|
||
|
|
COMMIT;"""))
|
||
|
|
|
||
|
|
|
||
|
|
def make_plan(source, target, kb_mapping, snapshot_sha256):
|
||
|
|
if target["database"] != TARGET_DATABASE or kb_mapping.get("tenant_id") != TARGET_TENANT:
|
||
|
|
raise ValueError("Unexpected acceptance database or knowledge tenant")
|
||
|
|
if any(source["grants"].values()):
|
||
|
|
raise ValueError("Explicit legacy grants changed from reviewed empty baseline; manual mapping required")
|
||
|
|
if source["resources"]["digitalhuman"] or source["memory_count"]:
|
||
|
|
raise ValueError("Digital human or memory source resources require additional mappings")
|
||
|
|
if source["roles"] != target["organization"]["source_role_definitions"]:
|
||
|
|
raise ValueError("Source role definitions differ from the durable identity snapshot")
|
||
|
|
roles = []
|
||
|
|
for role in source["roles"]:
|
||
|
|
if role["organization_id"] != 1:
|
||
|
|
raise ValueError("Cross-organization role")
|
||
|
|
if role["role"] == "admin":
|
||
|
|
if role["permissions"] != ["*"]:
|
||
|
|
raise ValueError("Administrator definition changed")
|
||
|
|
continue
|
||
|
|
permissions, notes = role_permissions(role)
|
||
|
|
roles.append({"source_role": role["role"], "id": stable(role["role"]), "name": "迁移 / " + role["label"],
|
||
|
|
"permissions": permissions, "source_permissions": role["permissions"], "review_notes": notes})
|
||
|
|
roles.append({"source_role": "historical-owner", "id": stable("historical-owner"), "name": "迁移 / 历史停用账户",
|
||
|
|
"permissions": [], "source_permissions": [], "review_notes": ["missing_source_identity_denied_all"]})
|
||
|
|
role_by_name = {r["source_role"]: r for r in roles}
|
||
|
|
source_users = {u["id"]: u for u in source["users"]}
|
||
|
|
assignments, identity_map = [], {}
|
||
|
|
for identity in target["identities"]:
|
||
|
|
uid, sid = identity["target_user_id"], identity["source_user_id"]
|
||
|
|
identity_map[sid] = uid
|
||
|
|
if sid == 1:
|
||
|
|
if uid != ADMIN_TARGET_ID:
|
||
|
|
raise ValueError("Pinned administrator mapping changed")
|
||
|
|
continue
|
||
|
|
memberships = [m for m in identity["memberships"] if m["deleted_at"] is None]
|
||
|
|
if (identity["user_tenant"] != TARGET_TENANT or identity["is_active"] is not False
|
||
|
|
or identity["is_system_admin"] is not False or identity["can_access_all_tenants"] is not False
|
||
|
|
or identity["user_deleted"] is not None or memberships != [{"tenant_id": TARGET_TENANT, "role": "contributor", "status": "suspended", "deleted_at": None}]):
|
||
|
|
raise ValueError("Legacy non-admin account must remain isolated, inactive and suspended")
|
||
|
|
original = source_users.get(sid)
|
||
|
|
if original:
|
||
|
|
if original["organization_id"] != 1 or original["enabled"] != 1:
|
||
|
|
raise ValueError("Disabled or cross-organization source user cannot acquire a role")
|
||
|
|
source_role = original["role"]
|
||
|
|
if identity["source_snapshot"].get("role") != source_role:
|
||
|
|
raise ValueError("Legacy identity role differs from durable mapping")
|
||
|
|
elif identity["source_snapshot"].get("tombstone") is True:
|
||
|
|
source_role = "historical-owner"
|
||
|
|
else:
|
||
|
|
raise ValueError("Missing legacy account without a reviewed tombstone")
|
||
|
|
assignments.append({"source_user_id": sid, "user_id": uid, "role_id": role_by_name[source_role]["id"]})
|
||
|
|
if {u["id"] for u in source["users"] if u["enabled"] == 1} - set(identity_map):
|
||
|
|
raise ValueError("An enabled legacy user lacks a durable identity mapping")
|
||
|
|
native = {(r["kind"], r["id"]): r for r in target["resources"]}
|
||
|
|
mappings = {(r["source_kind"], r["source_id"]): r for r in target["asset_mappings"]}
|
||
|
|
mappings.update({("canvas", r["source_id"]): r for r in target["flow_mappings"]})
|
||
|
|
publications = []
|
||
|
|
for table, source_kind, kind in (("agent", "agent", "agents"), ("agentskill", "skill", "skills"),
|
||
|
|
("canvas", "canvas", "flows"), ("knowledgebase", "knowledgebase", "knowledge")):
|
||
|
|
for row in source["resources"][table]:
|
||
|
|
if row["organization_id"] != 1 or row["visibility"] not in ("private", "global"):
|
||
|
|
raise ValueError("Unsupported resource organization or visibility")
|
||
|
|
if kind == "knowledge":
|
||
|
|
rid = kb_mapping["knowledge_bases"][str(row["id"])]["id"]
|
||
|
|
else:
|
||
|
|
mapping = mappings[(source_kind, row["id"])]
|
||
|
|
if mapping["source_visibility"] != row["visibility"]:
|
||
|
|
raise ValueError("Resource visibility differs from durable source mapping")
|
||
|
|
rid = mapping["target_id"]
|
||
|
|
actual = native[(kind, rid)]
|
||
|
|
if kind != "knowledge" and actual["owner_id"] != identity_map[row["owner_user_id"] or 1]:
|
||
|
|
raise ValueError("Resource owner differs from identity mapping")
|
||
|
|
enabled = row.get("status", "active") == "active" and bool(row.get("enabled", True))
|
||
|
|
if kind in ("agents", "skills") and (actual["visibility"] != row["visibility"] or actual["enabled"] != enabled):
|
||
|
|
raise ValueError("Native resource visibility or enablement has changed; review before import")
|
||
|
|
publications.append({"kind": kind, "resource_id": rid, "global": row["visibility"] == "global" and enabled,
|
||
|
|
"source_kind": source_kind, "source_id": row["id"], "expected_native": actual})
|
||
|
|
plan = {"schema_version": 1, "source": SOURCE, "target_database": TARGET_DATABASE, "tenant_id": TARGET_TENANT,
|
||
|
|
"source_snapshot_sha256": snapshot_sha256, "roles": roles, "assignments": assignments,
|
||
|
|
"publications": publications, "grants": [], "source_grant_counts": {t: len(rows) for t, rows in source["grants"].items()},
|
||
|
|
"source_role_definitions": source["roles"],
|
||
|
|
"prerequisites": (["apply_migration_000196"] if set(ACCESS_TABLES) - set(target["access_tables"]) else [])
|
||
|
|
+ ([] if target.get("content_permissions_ready") else ["apply_migration_000207"]),
|
||
|
|
"activation_blockers": ["verify_shared_knowledge_write_isolation_and_export_denial", "review_unmapped_legacy_logs_runs_usage_permissions", "reset_credentials_and_review_placeholder_email"],
|
||
|
|
"policy": "insert_only_no_activation_no_admin_changes_no_inferred_grants_preserve_native_edits"}
|
||
|
|
plan["plan_sha256"] = digest(plan)
|
||
|
|
return plan
|
||
|
|
|
||
|
|
|
||
|
|
def migration_sql(plan, commit=False):
|
||
|
|
if (plan["target_database"] != TARGET_DATABASE or plan["tenant_id"] != TARGET_TENANT
|
||
|
|
or plan["grants"] or plan["plan_sha256"] != digest({k: v for k, v in plan.items() if k != "plan_sha256"})):
|
||
|
|
raise ValueError("Plan scope or integrity mismatch")
|
||
|
|
payload = "'" + canonical(plan).replace("'", "''") + "'::jsonb"
|
||
|
|
return f"""-- Review plan SHA256: {plan['plan_sha256']}
|
||
|
|
-- This file is a rehearsal by default. No source database is accessed.
|
||
|
|
BEGIN ISOLATION LEVEL SERIALIZABLE;
|
||
|
|
SET LOCAL statement_timeout='30s';
|
||
|
|
SET LOCAL lock_timeout='5s';
|
||
|
|
SET LOCAL standard_conforming_strings=on;
|
||
|
|
DO $scope$ BEGIN
|
||
|
|
IF current_database() <> '{TARGET_DATABASE}' THEN RAISE EXCEPTION 'Wrong migration database'; END IF;
|
||
|
|
IF NOT EXISTS(SELECT 1 FROM tenants WHERE id={TARGET_TENANT} AND status='active' AND deleted_at IS NULL)
|
||
|
|
THEN RAISE EXCEPTION 'Missing active acceptance tenant'; END IF;
|
||
|
|
IF to_regclass('workbench_access_grants') IS NULL THEN RAISE EXCEPTION 'Apply schema migration 000196 first'; END IF;
|
||
|
|
IF to_regclass('workbench_content_permission_migrations') IS NULL THEN RAISE EXCEPTION 'Apply schema migration 000207 first'; END IF;
|
||
|
|
END $scope$;
|
||
|
|
SELECT pg_advisory_xact_lock({TARGET_TENANT},11905);
|
||
|
|
LOCK TABLE users,tenant_members,workbench_access_roles,workbench_access_assignments,
|
||
|
|
workbench_access_publications,workbench_access_grants IN SHARE ROW EXCLUSIVE MODE;
|
||
|
|
LOCK TABLE custom_agents,tenant_skill_catalog,knowledge_bases,workbench_resources IN SHARE MODE;
|
||
|
|
CREATE TEMP TABLE _access_users_before ON COMMIT DROP AS SELECT * FROM users;
|
||
|
|
CREATE TEMP TABLE _access_members_before ON COMMIT DROP AS SELECT * FROM tenant_members;
|
||
|
|
CREATE TEMP TABLE _access_roles_before ON COMMIT DROP AS SELECT * FROM workbench_access_roles;
|
||
|
|
CREATE TEMP TABLE _access_assignments_before ON COMMIT DROP AS SELECT * FROM workbench_access_assignments;
|
||
|
|
CREATE TEMP TABLE _access_publications_before ON COMMIT DROP AS SELECT * FROM workbench_access_publications;
|
||
|
|
CREATE TEMP TABLE _access_grants_before ON COMMIT DROP AS SELECT * FROM workbench_access_grants;
|
||
|
|
CREATE TABLE IF NOT EXISTS {LEDGER} (
|
||
|
|
source_system text NOT NULL,target_tenant_id bigint NOT NULL,entity_key text NOT NULL,
|
||
|
|
source_spec jsonb NOT NULL,plan_sha256 varchar(64) NOT NULL,
|
||
|
|
created_at timestamptz NOT NULL DEFAULT now(),PRIMARY KEY(source_system,target_tenant_id,entity_key)
|
||
|
|
);
|
||
|
|
DO $migrate$
|
||
|
|
DECLARE plan jsonb := {payload}; item jsonb; spec jsonb; saved jsonb; key text;
|
||
|
|
BEGIN
|
||
|
|
IF NOT EXISTS(SELECT 1 FROM yawan_legacy_organization_mappings WHERE source_system='{SOURCE}'
|
||
|
|
AND source_organization_id=1 AND target_tenant_id={TARGET_TENANT}
|
||
|
|
AND source_role_definitions=plan->'source_role_definitions')
|
||
|
|
THEN RAISE EXCEPTION 'Source role provenance changed'; END IF;
|
||
|
|
IF NOT EXISTS(SELECT 1 FROM yawan_legacy_identity_mappings WHERE source_system='{SOURCE}'
|
||
|
|
AND source_user_id=1 AND target_tenant_id={TARGET_TENANT} AND target_user_id='{ADMIN_TARGET_ID}')
|
||
|
|
THEN RAISE EXCEPTION 'Pinned administrator mapping changed'; END IF;
|
||
|
|
FOR item IN SELECT value FROM jsonb_array_elements(plan->'assignments') LOOP
|
||
|
|
IF NOT EXISTS(SELECT 1 FROM users u JOIN tenant_members m ON m.user_id=u.id
|
||
|
|
JOIN yawan_legacy_identity_mappings l ON l.target_user_id=u.id AND l.target_tenant_id={TARGET_TENANT}
|
||
|
|
WHERE u.id=item->>'user_id' AND u.tenant_id={TARGET_TENANT} AND NOT u.is_active AND NOT u.is_system_admin
|
||
|
|
AND NOT u.can_access_all_tenants AND u.deleted_at IS NULL AND m.tenant_id={TARGET_TENANT}
|
||
|
|
AND m.role='contributor' AND m.status='suspended' AND m.deleted_at IS NULL
|
||
|
|
AND l.source_user_id=(item->>'source_user_id')::bigint
|
||
|
|
AND NOT EXISTS(SELECT 1 FROM tenant_members other WHERE other.user_id=u.id AND other.tenant_id<>{TARGET_TENANT} AND other.deleted_at IS NULL))
|
||
|
|
THEN RAISE EXCEPTION 'Migrated account must remain suspended and isolated'; END IF;
|
||
|
|
END LOOP;
|
||
|
|
FOR item IN SELECT value FROM jsonb_array_elements(plan->'roles') LOOP
|
||
|
|
key := 'role/' || (item->>'source_role'); spec := item;
|
||
|
|
SELECT source_spec INTO saved FROM {LEDGER} WHERE source_system='{SOURCE}' AND target_tenant_id={TARGET_TENANT} AND entity_key=key;
|
||
|
|
IF FOUND THEN
|
||
|
|
IF saved IS DISTINCT FROM spec AND NOT EXISTS(
|
||
|
|
SELECT 1 FROM workbench_content_permission_migrations m,jsonb_array_elements(m.changes) change
|
||
|
|
WHERE m.version=207 AND change->>'role_id'=item->>'id'
|
||
|
|
AND (change->>'tenant_id')::bigint={TARGET_TENANT}
|
||
|
|
AND change->'legacy_source_spec'=saved
|
||
|
|
AND (saved-'permissions'-'review_notes')=(spec-'permissions'-'review_notes')
|
||
|
|
AND change->'mapped_permissions'=spec->'permissions')
|
||
|
|
THEN RAISE EXCEPTION 'Role migration provenance conflict'; END IF;
|
||
|
|
CONTINUE;
|
||
|
|
END IF;
|
||
|
|
IF EXISTS(SELECT 1 FROM workbench_access_roles WHERE id=item->>'id' OR (tenant_id={TARGET_TENANT} AND name=item->>'name'))
|
||
|
|
THEN RAISE EXCEPTION 'Native role collision'; END IF;
|
||
|
|
INSERT INTO workbench_access_roles(id,tenant_id,name,description,permissions,version)
|
||
|
|
VALUES(item->>'id',{TARGET_TENANT},item->>'name','旧系统权限保守迁移；原账户保持停用，启用前仍需审核。',item->'permissions',1);
|
||
|
|
INSERT INTO {LEDGER}(source_system,target_tenant_id,entity_key,source_spec,plan_sha256)
|
||
|
|
VALUES('{SOURCE}',{TARGET_TENANT},key,spec,plan->>'plan_sha256');
|
||
|
|
END LOOP;
|
||
|
|
FOR item IN SELECT value FROM jsonb_array_elements(plan->'assignments') LOOP
|
||
|
|
key := 'assignment/' || (item->>'user_id'); spec := item;
|
||
|
|
SELECT source_spec INTO saved FROM {LEDGER} WHERE source_system='{SOURCE}' AND target_tenant_id={TARGET_TENANT} AND entity_key=key;
|
||
|
|
IF FOUND THEN
|
||
|
|
IF saved IS DISTINCT FROM spec THEN RAISE EXCEPTION 'Assignment migration provenance conflict'; END IF;
|
||
|
|
CONTINUE;
|
||
|
|
END IF;
|
||
|
|
IF EXISTS(SELECT 1 FROM workbench_access_assignments WHERE tenant_id={TARGET_TENANT} AND user_id=item->>'user_id')
|
||
|
|
THEN RAISE EXCEPTION 'Native assignment collision'; END IF;
|
||
|
|
INSERT INTO workbench_access_assignments(tenant_id,user_id,role_id) VALUES({TARGET_TENANT},item->>'user_id',item->>'role_id');
|
||
|
|
INSERT INTO {LEDGER}(source_system,target_tenant_id,entity_key,source_spec,plan_sha256)
|
||
|
|
VALUES('{SOURCE}',{TARGET_TENANT},key,spec,plan->>'plan_sha256');
|
||
|
|
END LOOP;
|
||
|
|
FOR item IN SELECT value FROM jsonb_array_elements(plan->'publications') LOOP
|
||
|
|
key := 'publication/' || (item->>'kind') || '/' || (item->>'resource_id'); spec := item;
|
||
|
|
SELECT source_spec INTO saved FROM {LEDGER} WHERE source_system='{SOURCE}' AND target_tenant_id={TARGET_TENANT} AND entity_key=key;
|
||
|
|
IF FOUND THEN
|
||
|
|
IF saved IS DISTINCT FROM spec THEN RAISE EXCEPTION 'Publication migration provenance conflict'; END IF;
|
||
|
|
CONTINUE;
|
||
|
|
END IF;
|
||
|
|
IF EXISTS(SELECT 1 FROM workbench_access_publications WHERE tenant_id={TARGET_TENANT} AND kind=item->>'kind' AND resource_id=item->>'resource_id')
|
||
|
|
OR EXISTS(SELECT 1 FROM workbench_access_grants WHERE tenant_id={TARGET_TENANT} AND kind=item->>'kind' AND resource_id=item->>'resource_id')
|
||
|
|
THEN RAISE EXCEPTION 'Native resource authorization already exists'; END IF;
|
||
|
|
IF NOT EXISTS(SELECT 1 FROM (
|
||
|
|
SELECT 'agents' AS kind,id,tenant_id,created_by AS owner_id,visibility,enabled FROM custom_agents WHERE deleted_at IS NULL
|
||
|
|
UNION ALL SELECT 'skills',id,tenant_id,created_by,visibility,enabled FROM tenant_skill_catalog WHERE instruction_only=true AND deleted_at IS NULL
|
||
|
|
UNION ALL SELECT 'knowledge',id,tenant_id,creator_id,'global',true FROM knowledge_bases WHERE deleted_at IS NULL AND is_temporary=false
|
||
|
|
UNION ALL SELECT 'flows',id,tenant_id,owner_id,'private',true FROM workbench_resources WHERE kind='flows') r
|
||
|
|
WHERE to_jsonb(r)=item->'expected_native')
|
||
|
|
THEN RAISE EXCEPTION 'Native resource changed since plan'; END IF;
|
||
|
|
INSERT INTO workbench_access_publications(tenant_id,kind,resource_id,global)
|
||
|
|
VALUES({TARGET_TENANT},item->>'kind',item->>'resource_id',(item->>'global')::boolean);
|
||
|
|
INSERT INTO {LEDGER}(source_system,target_tenant_id,entity_key,source_spec,plan_sha256)
|
||
|
|
VALUES('{SOURCE}',{TARGET_TENANT},key,spec,plan->>'plan_sha256');
|
||
|
|
END LOOP;
|
||
|
|
END $migrate$;
|
||
|
|
DO $preservation$ BEGIN
|
||
|
|
IF EXISTS(SELECT 1 FROM _access_users_before b FULL JOIN users a ON a.id=b.id WHERE to_jsonb(a) IS DISTINCT FROM to_jsonb(b))
|
||
|
|
OR EXISTS(SELECT 1 FROM _access_members_before b FULL JOIN tenant_members a ON a.id=b.id WHERE to_jsonb(a) IS DISTINCT FROM to_jsonb(b))
|
||
|
|
OR EXISTS(SELECT 1 FROM _access_grants_before b FULL JOIN workbench_access_grants a USING(tenant_id,kind,resource_id,user_id) WHERE to_jsonb(a) IS DISTINCT FROM to_jsonb(b))
|
||
|
|
THEN RAISE EXCEPTION 'Existing users, memberships or grants changed'; END IF;
|
||
|
|
IF EXISTS(SELECT 1 FROM _access_roles_before b LEFT JOIN workbench_access_roles a ON a.id=b.id WHERE to_jsonb(a) IS DISTINCT FROM to_jsonb(b))
|
||
|
|
OR EXISTS(SELECT 1 FROM _access_assignments_before b LEFT JOIN workbench_access_assignments a USING(tenant_id,user_id) WHERE to_jsonb(a) IS DISTINCT FROM to_jsonb(b))
|
||
|
|
OR EXISTS(SELECT 1 FROM _access_publications_before b LEFT JOIN workbench_access_publications a USING(tenant_id,kind,resource_id) WHERE to_jsonb(a) IS DISTINCT FROM to_jsonb(b))
|
||
|
|
THEN RAISE EXCEPTION 'Existing native access configuration changed'; END IF;
|
||
|
|
END $preservation$;
|
||
|
|
SELECT json_build_object('tenant_id',{TARGET_TENANT},'mapping_rows',count(*),'plan_sha256','{plan['plan_sha256']}')
|
||
|
|
FROM {LEDGER} WHERE source_system='{SOURCE}' AND target_tenant_id={TARGET_TENANT};
|
||
|
|
{'COMMIT' if commit else 'ROLLBACK'};
|
||
|
|
"""
|
||
|
|
|
||
|
|
|
||
|
|
def main():
|
||
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
||
|
|
parser.add_argument("--sqlite", type=Path, required=True)
|
||
|
|
parser.add_argument("--kb-mapping", type=Path, required=True)
|
||
|
|
parser.add_argument("--out", type=Path, required=True)
|
||
|
|
parser.add_argument("--commit-sql", action="store_true", help="Emit COMMIT SQL only; still never execute it")
|
||
|
|
args = parser.parse_args()
|
||
|
|
sql_path = args.out.with_suffix(".sql")
|
||
|
|
if args.out.exists() or sql_path.exists():
|
||
|
|
parser.error("Output exists; choose a new review artifact name")
|
||
|
|
with consistent_snapshot(args.sqlite) as (db, info):
|
||
|
|
source = read_source(db)
|
||
|
|
if info["source_mutated"]:
|
||
|
|
raise ValueError("Source changed during snapshot")
|
||
|
|
target = read_target()
|
||
|
|
plan = make_plan(source, target, json.loads(args.kb_mapping.read_text()), info["sha256"])
|
||
|
|
with args.out.open("x", encoding="utf-8") as output:
|
||
|
|
json.dump(plan, output, ensure_ascii=False, indent=2)
|
||
|
|
with sql_path.open("x", encoding="utf-8") as output:
|
||
|
|
output.write(migration_sql(plan, args.commit_sql))
|
||
|
|
print(json.dumps({"roles": len(plan["roles"]), "assignments": len(plan["assignments"]),
|
||
|
|
"publications": len(plan["publications"]), "global": sum(p["global"] for p in plan["publications"]),
|
||
|
|
"grants": 0, "writes_performed": False, "source_mutated": info["source_mutated"],
|
||
|
|
"prerequisites": plan["prerequisites"], "plan_sha256": plan["plan_sha256"]}))
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
main()
|