#!/usr/bin/env python3 """Verify historical workbench imports through pinned local acceptance HTTP. Run only after the business import and read-only protection deployment finish. Reports contain only boolean checks and counts, never resource or login bodies. """ from __future__ import annotations import argparse from collections import Counter import json import secrets import urllib.error import urllib.request import uuid from pathlib import Path from apply_weknora_business import target_snapshot from apply_weknora_identities import sql_run from weknora_business_etl import LEDGER, RESOURCE_KINDS, SOURCE from weknora_category_etl import literal, save_json from weknora_identity_etl import TARGET_DATABASE, TARGET_TENANT BASE_URL = 'http://127.0.0.1:18080' ROOT = '/api/v1/workbench/resources/' EXPECTED_COUNTS = {'canvas': 3, 'trainingmap': 8, 'campaign': 2} ACTIONS = { 'flows': ('generate',), # Flows have no executable action in the native API. 'training-maps': ('generate', 'revise', 'generate-day', 'confirm-day', 'confirm'), 'content-tasks': ('decompose', 'generate', 'revise', 'accept', 'skip', 'review', 'compose'), } READ_ONLY_TEXT = '\u53ea\u8bfb\u5386\u53f2\u8d44\u6599' def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--credentials', type=Path, required=True) parser.add_argument('--out', type=Path, required=True) args = parser.parse_args() if args.out.exists(): parser.error('Report exists') opener = urllib.request.build_opener(urllib.request.ProxyHandler({})) checks, counts = {}, Counter() before, fixture_user, fixture_tenant = None, None, None def check(name, passed): checks[name] = checks.get(name, True) and bool(passed) counts[name] += 1 return bool(passed) def require(name, passed): if not check(name, passed): raise RuntimeError('Acceptance precondition failed') def request(path, token=None, data=None, method=None, tenant=TARGET_TENANT): headers = {'Content-Type': 'application/json'} if tenant is not None: headers['X-Tenant-ID'] = str(tenant) if token: headers['Authorization'] = 'Bearer ' + token body = json.dumps(data, ensure_ascii=False).encode() if data is not None else None if body is not None and len(body) > 2 * 1024 * 1024: raise RuntimeError('Verification payload exceeds the native request limit') req = urllib.request.Request(BASE_URL + path, headers=headers, data=body, method=method) counts['http_requests'] += 1 try: response = opener.open(req, timeout=30) except urllib.error.HTTPError as error: response = error with response: payload = json.load(response) if not isinstance(payload, dict): raise RuntimeError('Unexpected acceptance response shape') return response.status, payload def listed(kind, token): result = {} for page in range(1, 1001): status, body = request(ROOT + kind + f'?page={page}&page_size=100', token) if status != 200 or not isinstance(body.get('data'), list): return False, result rows, total = body['data'], body.get('total') if type(total) is not int or total < 0: return False, result old_count = len(result) for row in rows: result[row['id']] = row if len(result) >= total: return len(result) == total, result if len(result) == old_count: return False, result return False, result def denied(name, path, token, data=None, method=None, tenant=TARGET_TENANT, statuses=(403, 404), readonly=False): status, body = request(path, token, data, method, tenant) passed = status in statuses and (not readonly or READ_ONLY_TEXT in body.get('message', '')) check(name, passed) # A successful mutation ends the run immediately; never attempt repairs. if method in ('PUT', 'DELETE', 'POST') and 200 <= status < 300: raise RuntimeError('A protected mutation unexpectedly succeeded') try: before = target_snapshot() require('pinned_acceptance_database', before['database'] == TARGET_DATABASE) expected = json.loads(sql_run( f"SELECT coalesce(json_agg(json_build_object('kind',source_kind,'id',target_id," f"'native',native_snapshot) ORDER BY source_kind,source_id),'[]') FROM {LEDGER} " f"WHERE source_system={literal(SOURCE)} AND source_organization_id=1 " f"AND target_tenant_id={TARGET_TENANT} AND source_kind IN ('canvas','trainingmap','campaign');")) require('exact_13_migrated_resources_in_ledger', Counter(item['kind'] for item in expected) == EXPECTED_COUNTS and len({item['id'] for item in expected}) == 13) for source_kind, count in EXPECTED_COUNTS.items(): counts['migrated_' + RESOURCE_KINDS[source_kind]] = count require('all_13_native_snapshots_are_read_only', all( item['native']['data'].get('legacy_import', {}).get('read_only') is True and item['native']['tenant_id'] == TARGET_TENANT and item['native']['id'] == item['id'] and item['native']['kind'] == RESOURCE_KINDS[item['kind']] for item in expected)) credentials = json.loads(args.credentials.read_text()) status, auth = request('/api/v1/auth/login', data={k: credentials[k] for k in ('email', 'password')}) require('administrator_login_succeeded', status == 200 and bool(auth.get('token'))) token, administrator_id = auth['token'], auth.get('user', {}).get('id') require('administrator_identity_available', bool(administrator_id)) counts['migrated_distinct_owners'] = len({item['native']['owner_id'] for item in expected}) counts['migrated_administrator_owned_resources'] = sum( item['native']['owner_id'] == administrator_id for item in expected) counts['migrated_other_owner_resources'] = len(expected) - counts['migrated_administrator_owned_resources'] for kind in RESOURCE_KINDS.values(): ok, resources = listed(kind, token) selected = [item for item in expected if item['native']['kind'] == kind] require('administrator_lists_all_migrated_resources', ok and all(item['id'] in resources for item in selected)) denied('anonymous_list_denied', ROOT + kind, None, statuses=(401, 403)) for item in expected: native = item['native'] path = ROOT + native['kind'] + '/' + item['id'] status, body = request(path, token) current = body.get('data', {}) require('native_data_owner_status_and_version_match', status == 200 and all( current.get(field) == native[field] for field in ('id', 'tenant_id', 'owner_id', 'kind', 'name', 'status', 'version', 'data'))) status, body = request(path + '/jobs?page_size=100', token) require('imported_resources_have_no_jobs', status == 200 and body.get('data') == [] and body.get('total') == 0) for suffix in ('', '/jobs?page_size=100'): denied('anonymous_resource_and_jobs_read_denied', path + suffix, None, statuses=(401, 403)) payload = {key: native[key] for key in ('name', 'data', 'version')} owned = native['owner_id'] == administrator_id # PUT checks historical protection before locking. DELETE and supported # actions lock by owner first, even for a tenant administrator. denied('all_13_updates_rejected_as_read_only', path, token, payload, 'PUT', statuses=(400,), readonly=True) for detach, label in (('', 'deletes'), ('?detach=true', 'detaching_deletes')): denied(('owned_' + label + '_rejected_as_read_only') if owned else ('other_owner_' + label + '_rejected_by_ownership'), path + detach, token, method='DELETE', statuses=(400,) if owned else (403, 404), readonly=owned) for action in ACTIONS[native['kind']]: parameters = {'version': native['version'], 'day': 1} if action == 'revise': parameters['advice'] = 'Verification only; preserve historical content.' segments = native['data'].get('segments', []) if segments: parameters['segment_id'] = segments[0]['id'] is_flow = native['kind'] == 'flows' action_check = 'flow_actions_unsupported' if is_flow else ( 'owned_supported_actions_rejected_as_read_only' if owned else 'other_owner_supported_actions_rejected_by_ownership') denied(action_check, path + '/actions/' + action, token, parameters, 'POST', statuses=(400,) if is_flow or owned else (403, 404), readonly=owned and not is_flow) denied('anonymous_update_denied', path, None, payload, 'PUT', statuses=(401, 403)) denied('anonymous_delete_denied', path, None, method='DELETE', statuses=(401, 403)) denied('anonymous_action_denied', path + '/actions/generate', None, {'version': native['version']}, 'POST', statuses=(401, 403)) require('resources_and_execution_unchanged_after_admin_checks', before == target_snapshot()) suffix = secrets.token_hex(6) password = secrets.token_urlsafe(18) + 'aA1!' email = 'business-check-' + suffix + '@migration.invalid' existing_tenants = set(json.loads(sql_run("SELECT coalesce(json_agg(id),'[]') FROM tenants;"))) status, body = request('/api/v1/system/admin/users/create', token, { 'username': 'business-check-' + suffix, 'email': email, 'password': password}) candidate = body.get('user', {}).get('id') require('temporary_user_created', status == 201 and candidate and str(uuid.UUID(candidate)) == candidate) fixture_user = candidate home = int(sql_run(f"SELECT tenant_id FROM users WHERE id={literal(fixture_user)};")) if home and home not in existing_tenants: fixture_tenant = home require('temporary_user_home_is_new_or_tenantless', home == 0 or fixture_tenant == home) sql_run(f"INSERT INTO tenant_members(user_id,tenant_id,role,status) VALUES " f"({literal(fixture_user)},{TARGET_TENANT},'contributor','active');") status, auth = request('/api/v1/auth/login', data={'email': email, 'password': password}) require('contributor_login_succeeded', status == 200 and bool(auth.get('token'))) member_token = auth['token'] fixture_scope = json.loads(sql_run(f"SELECT json_build_object(" f"'is_admin',is_system_admin,'all_tenants',can_access_all_tenants," f"'memberships',(SELECT count(*) FROM tenant_members WHERE user_id=u.id AND deleted_at IS NULL)," f"'has_personal_membership',EXISTS(SELECT 1 FROM tenant_members WHERE user_id=u.id " f"AND tenant_id={fixture_tenant or 0} AND role='owner' AND status='active' AND deleted_at IS NULL)," f"'is_contributor',EXISTS(SELECT 1 FROM tenant_members WHERE user_id=u.id " f"AND tenant_id={TARGET_TENANT} AND role='contributor' AND status='active' AND deleted_at IS NULL)) " f"FROM users u WHERE id={literal(fixture_user)};")) require('temporary_user_is_isolated_contributor', fixture_scope == { 'is_admin': False, 'all_tenants': False, 'memberships': 1 + bool(fixture_tenant), 'has_personal_membership': bool(fixture_tenant), 'is_contributor': True}) for kind in RESOURCE_KINDS.values(): ok, resources = listed(kind, member_token) require('contributor_lists_no_other_owners_resources', ok and not resources) denied('cross_tenant_list_denied', ROOT + kind, member_token, tenant=10060) for item in expected: native = item['native'] path = ROOT + native['kind'] + '/' + item['id'] payload = {key: native[key] for key in ('name', 'data', 'version')} for suffix in ('', '/jobs?page_size=100'): denied('contributor_direct_resource_and_jobs_read_denied', path + suffix, member_token) denied('cross_tenant_direct_resource_and_jobs_read_denied', path + suffix, member_token, tenant=10060) denied('contributor_update_denied', path, member_token, payload, 'PUT') denied('contributor_delete_denied', path, member_token, method='DELETE') denied('cross_tenant_update_denied', path, member_token, payload, 'PUT', tenant=10060) denied('cross_tenant_delete_denied', path, member_token, method='DELETE', tenant=10060) denied('cross_tenant_action_denied', path + '/actions/generate', member_token, {'version': native['version']}, 'POST', tenant=10060) if native['kind'] != 'flows': denied('contributor_direct_action_denied', path + '/actions/generate', member_token, {'version': native['version']}, 'POST') else: denied('contributor_flow_action_unsupported', path + '/actions/generate', member_token, {'version': native['version']}, 'POST', statuses=(400,)) check('all_api_checks_completed', True) except Exception: # Transport/database exceptions can contain private paths or response data. check('all_api_checks_completed', False) finally: if fixture_user: try: disable_tenant = (f"UPDATE tenants SET status='inactive',updated_at=now() " f"WHERE id={fixture_tenant}; ") if fixture_tenant else '' sql_run(f"BEGIN; DELETE FROM tenant_members WHERE user_id={literal(fixture_user)}; " f"DELETE FROM auth_tokens WHERE user_id={literal(fixture_user)}; " f"UPDATE users SET is_active=false,updated_at=now() WHERE id={literal(fixture_user)}; " + disable_tenant + "COMMIT;") cleaned = sql_run(f"SELECT NOT is_active AND NOT EXISTS(SELECT 1 FROM tenant_members " f"WHERE user_id=u.id) AND NOT EXISTS(SELECT 1 FROM auth_tokens WHERE user_id=u.id) " f"FROM users u WHERE id={literal(fixture_user)};") == 't' check('temporary_user_deactivated_membership_and_tokens_removed', cleaned) if fixture_tenant: check('temporary_personal_tenant_deactivated', sql_run( f"SELECT status='inactive' FROM tenants WHERE id={fixture_tenant};") == 't') except Exception: check('temporary_user_deactivated_membership_and_tokens_removed', False) if before is not None: try: check('all_existing_resources_jobs_events_revisions_assets_and_knowledge_unchanged', before == target_snapshot()) except Exception: check('all_existing_resources_jobs_events_revisions_assets_and_knowledge_unchanged', False) result = {'passed': bool(checks) and all(checks.values()), 'checks': checks, 'counts': dict(counts), 'fixture_user_created': fixture_user is not None, 'fixture_personal_tenant_created': fixture_tenant is not None} save_json(args.out, result) print(json.dumps(result)) return 0 if result['passed'] else 1 if __name__ == '__main__': raise SystemExit(main())