#!/usr/bin/env python3 """Offline exact native-row/SQL/bind/key/cleanup verification for approved 2A only.""" import copy import hashlib import json import re import sys from pathlib import Path ROOT = Path(__file__).parent PRIVATE = ROOT / 'probe2a-private-evidence' SHA = 'c23f516cd9764e1eb6dec57f8ca5cb579c74f6c25008b25e6d242dfe1e14f78d' FIELDS = ['id', 'marker', 'dingtalk_userid', 'primary_department_uuid', 'dingtalk_sync_md5'] HEX_FIELDS = {'dingtalk_userid': 'userid_hex', 'primary_department_uuid': 'primary_hex', 'dingtalk_sync_md5': 'md5_hex'} def key(value): return None if value is None else (value.rstrip(' ').encode().hex().upper() or '#') def verify(plan, lanes): native_steps = {} summary = {} for lane, events in lanes.items(): grouped = {} for event in events: grouped.setdefault(event['kind'], []).append(event['data']) def one(kind): assert len(grouped[kind]) == 1 return grouped[kind][0] assert not {'stopped', 'skipped'} & grouped.keys() spec = plan['lanes'][lane] ddl = [x for x in [spec['create_table'], spec['create_unique_index'], spec['create_trigger']] if x] + spec['cleanup'] assert [x['SQL'] for x in grouped['DDL-start']] == [x['SQL'] for x in grouped['DDL-success']] == ddl for item in grouped['DDL-start']: assert item['sha256'] == hashlib.sha256(item['SQL'].encode()).hexdigest() assert all('OR REPLACE' not in sql and 'IF NOT EXISTS' not in sql for sql in ddl) assert one('collision-before') == one('collision-after') == {'table': [], 'index': [], 'trigger': []} assert one('business-before') == one('business-after') assert len(one('business-before')['columns']) == (77 if lane == 'B' else 70) assert one('business-before')['aggregate'][0]['total'] == (40930 if lane == 'B' else 2103) final_meta = one('metadata-created-final') assert [c['column_name'].lower() for c in final_meta['columns']] == FIELDS if lane == 'C': trigger = final_meta['triggers'] assert len(trigger) == 1 and trigger[0]['status'] == 'Y' assert trigger[0]['trigger_name'] == 'TRG_CDX_SS7_2A_CHAR' assert trigger[0]['trigger_body'].endswith(spec['create_trigger'].split(' FOR EACH ROW')[1]) assert 'UPDATE_TIME' not in trigger[0]['trigger_body'] assert any(i['index_name'] == 'UX_CDX_SS7_2A_USERID' and i['uniqueness'] == 'UNIQUE' for i in final_meta['indexes']) starts, results = grouped['DML-start'], grouped['DML-result'] assert len(starts) == len(results) == 33 expected = {} previous = [] native_steps[lane] = [] counts = [] for step, start, end in zip(spec['steps'], starts, results): bindings = [{'value': v, 'type': t} for v, t in zip(step['binds'], step['PDO_types'])] assert end['SQL'] == start['SQL'] == step['SQL'] assert end['label'] == start['label'] == step['label'] assert end['bindings'] == start['bindings'] == bindings assert start == {k: end[k] for k in start} assert end['before'] == previous and end['expect_reject'] == step['expect_reject'] assert end['ok'] == (not step['expect_reject']) and end['observed_hypothesis_match'] if end['ok']: if step['SQL'].startswith('INSERT'): columns = step['SQL'].split(' (', 1)[1].split(')', 1)[0].lower().split(',') row = dict(zip(columns, step['binds'])) row.setdefault('primary_department_uuid', None) row.setdefault('dingtalk_sync_md5', '') assert row['id'] not in expected expected[row['id']] = row else: columns = step['SQL'].split(' SET ')[1].split(' WHERE ')[0].split(',') pk = step['binds'][-1] assert pk in expected expected[pk].update({column.split('=')[0].lower(): value for column, value in zip(columns, step['binds'][:-1])}) for row in expected.values(): for field in ('primary_department_uuid', 'dingtalk_sync_md5'): if row[field] is not None: row[field] = row[field].rstrip(' ') assert end['rowCount'] == (0 if lane == 'B' and step['label'] == 'P12-CHAR-single-space' else 1) counts.append([step['label'], end['rowCount']]) else: assert end['after'] == previous and ('1062' if lane == 'B' else '-6602') in end['error']['message'] actual = end['after'] assert len(actual) == len(expected) <= 6 assert [row['id'] for row in actual] == sorted(expected) for row, check in zip(actual, end['byte_checks']): assert set(row) == set(FIELDS + list(HEX_FIELDS.values()) + ['userid_key']) assert {f: row[f] for f in FIELDS} == expected[row['id']] assert type(row['id']) is int and row['marker'] == 'CDXSS2A' assert check['id'] == row['id'] for field, alias in HEX_FIELDS.items(): value = row[field] encoded = value.encode() if value is not None else None expected_hex = encoded.hex().upper() if encoded else ('' if lane == 'B' and value == '' else None) assert row[alias] == expected_hex assert check[field] == {'native_type': 'NULL' if value is None else 'string', 'bytes': len(encoded) if encoded is not None else None, 'sha256': hashlib.sha256(encoded).hexdigest() if encoded is not None else None, 'HEX_exact': True} assert row['userid_key'] == check['key_expected'] == check['key_actual'] == key(row['dingtalk_userid']) assert check['key_exact'] native_steps[lane].append([{f: row[f] for f in FIELDS} for row in actual]) previous = actual assert one('own-final') == previous and len(previous) == 6 predicates = one('predicate-final') assert len(predicates) == len(spec['predicate_queries']) for query, actual in zip(spec['predicate_queries'], predicates): assert actual['SQL'] == query['SQL'] and actual['kind'] == query['kind'] assert actual['bindings'] == [{'value': query['binds'][0], 'type': 2}] if query['kind'] == 'key': wanted = [{'id': row['id']} for row in previous if row['userid_key'] == query['binds'][0]] assert actual['rows'] == wanted journal = one('read-journal') assert all(q['SQL'].startswith('SELECT ') and q['ok'] for q in journal) assert len(journal) == (86 if lane == 'B' else 122) <= 160 s = one('summary') assert s['completed'] and s['staff_metadata_aggregate_exact'] assert s['SELECT'] == len(journal) and s['DML_attempts'] == 33 and s['DDL_attempts'] == len(ddl) assert s['HTTP'] == s['business_table_writes'] == s['objects_remaining'] == 0 summary[lane] = {'attempts': 33, 'success': 27, 'unique_rejections': 6, 'DDL': len(ddl), 'SELECT': len(journal), 'objects_remaining': 0, 'Staff_metadata_aggregate_exact': True, 'noop_P12_rowCount': 0 if lane == 'B' else 1} assert native_steps['B'] == native_steps['C'] predicates = {lane: next(e['data'] for e in events if e['kind'] == 'predicate-final') for lane, events in lanes.items()} for kind in ('plain', 'key'): b = [q['rows'] for q in predicates['B'] if q['kind'] == kind] c = [q['rows'] for q in predicates['C'] if q['kind'] == kind] assert b == c return {'passed_2A_valid_domain': True, 'full_Staff_compatibility_passed': False, 'complete_native5_fields_each_step_equal': True, 'lanes': summary, 'ordinary_WHERE_case_gap_still_open': 'Final predicate data no longer contains A/a; first-probe live counterexample remains, no broader claim', 'residuals': ['invalid_ASCII_coercion_vs_CHECK', 'UINT64_native_double_precision', '2B_nullable_CLOB_time_not_tested'], 'Staff_business_DDL_DML_HTTP': 0} def main(): raw = (ROOT / 'probe2a-hex-char-proposal.json').read_bytes() assert hashlib.sha256(raw).hexdigest() == SHA plan = json.loads(raw) lanes = {lane: [json.loads(line) for line in (PRIVATE / (lane + '-probe2a.jsonl')).read_text().splitlines()] for lane in ('B', 'C')} assert all((PRIVATE / (lane + '-probe2a.stderr')).read_bytes() == b'' for lane in lanes) manifest_path = PRIVATE / 'manifest.json' if manifest_path.exists(): manifest = json.loads(manifest_path.read_text()) assert {e['path'] for e in manifest['files']} == {'B-probe2a.jsonl', 'B-probe2a.stderr', 'C-probe2a.jsonl', 'C-probe2a.stderr', 'gate.json'} for entry in manifest['files']: content = (PRIVATE / entry['path']).read_bytes() assert len(content) == entry['bytes'] and hashlib.sha256(content).hexdigest() == entry['sha256'] result = verify(plan, lanes) if '--test-mutations' in sys.argv: mutations = [ ('C', 'DML-result', 'P04-lower', lambda d: d.update(ok=False)), ('C', 'DML-result', 'P06-empty', lambda d: d['after'][-1].update(userid_key=None)), ('C', 'DML-result', 'P15-CHAR-tab', lambda d: d['after'][-1].update(dingtalk_sync_md5='A')), ('C', 'DML-result', 'P16-CHAR-NBSP-NUL', lambda d: d['after'][-1].update(dingtalk_sync_md5='A')), ('C', 'DML-result', 'P01-null1', lambda d: d['after'][0].update(dingtalk_sync_md5=' ' * 32)), ('B', 'DML-result', 'P17-CHAR-full-capacity', lambda d: d['after'][-1].update(primary_department_uuid='😀' * 35)), ('C', 'DML-result', 'P10-max-last-codepoint', lambda d: d['after'][-1].update(dingtalk_userid='😀' * 128)), ('B', 'DML-result', 'P28-userid-tab-tail-duplicate', lambda d: d.update(ok=True)), ('C', 'collision-after', None, lambda d: d.update(trigger=[{'trigger_name': 'TRG_CDX_SS7_2A_CHAR'}])), ('B', 'business-after', None, lambda d: d['aggregate'][0].update(total=40931)), ('C', 'DML-result', 'P19-CHAR-null-and-spaces', lambda d: d['bindings'][0].update(type=2)), ('C', 'DML-result', 'P12-CHAR-single-space', lambda d: d.update(rowCount=0)), ] for lane, kind, label, alter in mutations: changed = copy.deepcopy(lanes) for event in changed[lane]: if event['kind'] == kind and (label is None or event['data'].get('label') == label): alter(event['data']) try: verify(plan, changed) except AssertionError: continue raise AssertionError('Undetected mutation: ' + str((lane, kind, label))) result['mutations_rejected'] = len(mutations) print(json.dumps(result, ensure_ascii=False, indent=2)) if __name__ == '__main__': main()