"""Independent, strictly typed offline audit. No database or network access.""" import base64, copy, datetime, hashlib, json, re from pathlib import Path ROOT=Path(__file__).parent checks=0 def same(a,b): if type(a) is not type(b): return False if isinstance(a,dict): return list(a)==list(b) and all(same(a[k],b[k]) for k in a) if isinstance(a,list): return len(a)==len(b) and all(same(x,y) for x,y in zip(a,b)) return a==b def eq(a,b,label): global checks if not same(a,b): raise AssertionError(label) checks+=1 def read(p): return json.loads(p.read_text()) def partial(s): return {k:v for k,v in s.items() if k not in ('ai_users','config')} def protected(s,uuid): s=partial(s); s['activity']=[r for r in s['activity'] if r['uuid']!=uuid]; return s TIMES=('register_start','register_end','start_time','end_time','create_time') def response_row(row,lane): row=dict(row) if lane=='C': for k in TIMES: value=row.get(k) if not isinstance(value,str) or not re.fullmatch(r'\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}\.0{1,6}',value): continue seconds=value.split('.')[0] try: valid=datetime.datetime.strptime(seconds,'%Y-%m-%d %H:%M:%S') except ValueError: continue if valid.year>=1000: row[k]=seconds row['activity_status']='未开始' return row def dto(rows,method,lane,stage): if method=='info': row=response_row(rows[0],lane); row['participants']=0 return {'code':0,'message':'信息获取成功','data':row} if method=='lists': return {'code':0,'message':'请求成功','data':{'rows':[response_row(r,lane) for r in rows],'total':len(rows)}} if stage==8: return {'code':1,'message':'修改失败','data':[]} return {'code':0,'message':{'add':'新增活动成功','update':'编辑活动成功','able':'更改状态成功','delete':'删除活动成功'}[method],'data':[]} def raw_value(field,value,lane): if lane=='C' and field in TIMES: return value+'.000000' return value def sql_statements(native): return [r for r in native if 'sql' in r] def dml(native): return [r for r in sql_statements(native) if re.match(r'^\s*(INSERT|UPDATE|DELETE)\b',r['sql'])] summary={}; dtos={}; original_rows={} for lane in ('B','C'): d=ROOT/'private'/lane baseline=read(d/'baseline.json'); own=read(d/'owned.json'); uuid=own['uuid']; pre=read(d/'preflight.json'); proof=read(d/'tap-proof.json') eq(pre['marker_collision'],0,'Fresh unique marker'); eq(proof['before_first_HTTP_and_DML'],True,'Dual tap proven before first write') eq(proof['source'],partial(baseline),'Pre-write source fresh all scoped rows'); eq(read(d/'puregate.json')['puregate_checks'],39,'Actual pure gate checks') for role in ('store','ai'): p=proof['proof'][role] eq(p['native_class'],'ActPDO','Actual native tap'); eq(p['identity'][0]['db'],p['config_database'],'Actual scoped DB identity') eq(p['class'],'think\\db\\connector\\Mysql' if lane=='B' else 'app\\cores\\database\\Dm','Original DB driver') eq(len(proof['native']),4,'Four actual native pre-write proof reads') previous=partial(baseline); dtos[lane]={}; original_rows[lane]={} reads=len(proof['native']); inserts=updates=deletes=0 observer=len(pre['observer'])+len(read(d/'auth-proof.json')['observer'])+len(proof['observer']) for stage in range(1,12): label=f'{stage:02d}'; w=read(d/f'wire-{label}.json'); c=read(d/f'client-{label}.json'); r=read(d/f'response-{label}.json'); case=w['case']; method=case['method']; after=w['after'] eq(w['before'],previous,f'{lane}/{stage}: fresh full scoped continuity') eq(protected(after,uuid),partial(baseline),'All protected scoped full rows unchanged') eq('ai_users' in w['before'],False,'Intermediate full AI user hash not falsely cached'); eq('config' in after,False,'Intermediate full config hash omitted') eq(c['status'],200,'Client HTTP200'); eq(r['status'],200,'Original Response status200') eq(c['body'],r['body'],'Exact client vs original response bytes'); eq(w['body'],r['body'],'Exact server vs original response bytes') eq(hashlib.sha256(r['body'].encode()).hexdigest(),r['body_sha256'],'Original immutable body SHA') eq(json.loads(base64.b64decode(w['raw_request_base64'])),case['params'],'Actual raw JSON request') eq(w['warnings'],[],'Request-local PHP warnings'); eq(w.get('business_error'),None,'No source error'); eq(w.get('observer_error'),None,'No observation failure') eq(sorted(set(x['schema'] for x in w['native'])),['ai','store'],'Actual HTTP both native connections') ownrows=[r for r in after['activity'] if r['uuid']==uuid] expected=dto([r for r in after['activity'] if 'CDXACT11W2' in r['title']] if method=='lists' else ownrows,method,lane,stage) actual=json.loads(r['body']); eq(actual,expected,'Independent complete typed DTO'); dtos[lane][stage]=actual if method in ('add','update','able'): row=ownrows[0]; original_rows[lane][stage]=row for key,value in case['params'].items(): if key!='uuid': eq(str(row[key]),str(raw_value(key,value,lane)),'Physical exact request value and datetime precision') if stage!=1: original=[r for r in w['before']['activity'] if r['uuid']==uuid][0]; merged=dict(original) for key,value in case['allowed']['data'].items(): merged[key]=int(value) if type(row[key]) is int else value eq(row,merged,'Only exact approved own fields change') else: eq(row,own['full_created_row'],'Complete actual API creation journal'); eq(row['store_id'],10001,'Actual tenant') eq(row['create_by'],baseline['actor'][0]['name'],'Original creator function'); assert re.fullmatch(r'[A-Fa-f0-9]{8}(?:-[A-Fa-f0-9]{4}){3}-[A-Fa-f0-9]{12}',uuid) for field in TIMES: column=next(x for x in pre['metadata']['activity_columns'] if x['column_name'].lower()==field) eq(column['data_type'],'datetime' if lane=='B' else 'DATETIME','Actual time metadata') eq(column['datetime_precision'] if lane=='B' else column['data_scale'],0 if lane=='B' else 6.0,'Original physical datetime scale') native=sql_statements(w['native']); writes=dml(w['native']); reads+=len(native)-len(writes) eq(w['DML_attempts'],1 if case['allowed'] else 0,'All actual native DML attempt count'); eq(len(writes),w['DML_attempts'],'Native journal DML count') for x in writes: eq(x['schema'],'ai','Only exact AI table write'); eq(x['execute_parameters'],None,'Explicit native binds only'); eq(x['success'],True,'Original native execution success') eq(x['affected'],0 if stage==8 else 1,'Canonical no-op alone may affect zero') if stage==1: inserts+=1; columns=list(case['params'])+['create_by','uuid','create_time','store_id'] sql='INSERT INTO `yoshop_activity` (`'+'` , `'.join(columns)+'`) VALUES ('+' , '.join(':data__'+k for k in columns)+') ' eq(x['sql'],sql,'Exact original INSERT SQL bytes and field order') values=dict(case['params']); values.update(create_by=baseline['actor'][0]['name'],uuid=uuid,create_time=own['full_created_row']['create_time'].split('.')[0],store_id=10001) bind={':data__'+k:{'value':values[k],'type':1 if k in ('type_id','status','store_id') else 2,'php_type':'integer' if type(values[k]) is int else 'string','ok':True} for k in columns} eq(x['binds'],bind,'Complete original INSERT bind values, PHP and PDO types') ids=[z for z in w['native'] if z.get('layer')=='PDO.lastInsertId']; eq(len(ids),1,'Original lastInsertId once'); eq(ids[0]['schema'],'ai','Original identity same AI connection') eq(ids[0]['value'],str(own['id']),'Actual original generated ID matches observed PK'); eq(ids[0]['php_type'],'string','Native identity PHP type') elif method in ('update','able'): updates+=1 fields=['title','description','rules'] if stage==4 else ['status'] sql='UPDATE `yoshop_activity` SET '+','.join('`'+k+'`=:data__'+k for k in fields)+' WHERE `uuid` = :where_AND_uuid' if lane=='C' and method=='able': sql+=' AND `status` <> :where_AND_status' sql+=' '; eq(x['sql'],sql,'Exact actual source UPDATE with DM canonical predicate only') bind={':data__'+k:{'value':case['params'][k],'type':1 if k=='status' else 2,'php_type':'string','ok':True} for k in fields} bind[':where_AND_uuid']={'value':uuid,'type':2,'php_type':'string','ok':True} if lane=='C' and method=='able': bind[':where_AND_status']={'value':0,'type':1,'php_type':'integer','ok':True} eq(x['binds'],bind,'Exact UPDATE bind values and types') if stage==8: eq(ownrows[0]['status'],0,'Repeated canonical current status0'); eq(w['before'],w['after'],'Full repeated no-op source unchanged') else: deletes+=1; eq(ownrows,[],'Only new own row deleted'); lease=read(d/'delete-lease.json'); eq(lease['references'],[],'All-tenant registration reference0') eq(x['sql'],'DELETE FROM `yoshop_activity` WHERE `uuid` IN (:where_AND_uuid_in_1) ','Exact source DELETE SQL bytes') eq(x['binds'],{':where_AND_uuid_in_1':{'value':uuid,'type':2,'php_type':'string','ok':True}},'Exact only own DELETE binding') eq(lease['full_own_row']['id'],own['id'],'Deletion newly created PK'); eq(lease['full_own_row']['uuid'],uuid,'Deletion newly created UUID') eq(lease['full_own_row']['title'],'CDXACT11W2-U','Deletion exact own title'); eq(lease['full_own_row']['store_id'],10001,'Deletion exact tenant') observer+=len(w['observer']); previous=after # Client observer execution records are distinct only for first request and reference proof. first_before=read(d/'before-01.json')['observer']; first_result=read(d/'result-01.json')['observer'] eq(first_result[:len(first_before)],first_before,'Original first client observer prefix') observer+=len(first_result)+len(read(d/'before-10.json')['observer']) final=read(d/'final.json'); observer+=len(final['observer']) eq(previous,partial(baseline),'All scoped rows return exactly to original'); eq(final['source'],baseline,'Fresh final whole AI users/config and all scoped data exact') eq(inserts,1,'One API INSERT'); eq(updates,3,'Three UPDATE attempts'); eq(deletes,1,'One API DELETE') budget={p.stem.replace('budget-',''):int(p.read_text()) for p in d.glob('budget-*.json')} eq(budget['observer'],observer,'All independently counted observer executions'); eq(observer,288,'Observer budget exact') eq(budget['business-read'],reads,'Every native read incl metadata and proof counted'); eq(budget['total-read'],observer+reads,'Unified total exact') assert budget['total-read']<=800 for k,v in [('HTTP',11),('DML-insert',1),('DML-update',3),('DML-delete',1)]: eq(budget[k],v,'Exact approved request/write budget') eq(read(d/'cleanup.json')['own_port_closed'],True,'Own port closed'); eq(read(d/'cleanup.json')['original_parent_alive'],True,'Original parent preserved') eq(read(d/'cleanup-source.json')['removed_own_copied_source_files'],959,'Own source copy removed'); eq(read(d/'resource-proof.json')['source_mapping_sha256'],'f26f6dd15e86f3b34e23cde8413b660683cc5656f542b85973581c441c747738','Root actual APP959 exact') summary[lane]={'HTTP':11,'native_INSERT':inserts,'native_UPDATE_attempts':updates,'native_DELETE':deletes,'observer_SELECT':observer,'native_read_including_metadata_and_proof':reads,'unified_readonly_total':budget['total-read'],'full_scope_before_after_each_HTTP':True,'whole_AIuser_config_hash_frequency':'start/end only','final_all_state_exact':True,'canonical_repeat_code':1,'canonical_repeat_affected':0,'resource_include_records':read(d/'resource-proof.json')['actual_non_tool_include_records_checked']} cross=0 for stage in range(1,12): b=dtos['B'][stage]; c=copy.deepcopy(dtos['C'][stage]) if stage in (2,3,5,7,9): br=b['data']['rows'][0] if stage==3 else b['data']; cr=c['data']['rows'][0] if stage==3 else c['data'] for field in ('id','uuid','create_by','create_time'): cr[field]=br[field] eq(c,b,'Cross-DB complete DTO exact after four proven environment fields mapped'); cross+=1 for a,b in [(1,1.0),([],{}),(None,''),(0,False),({'x':1},{'x':1.0}),({'a':1,'b':2},{'b':2,'a':1}),('2099-02-01 00:00:00','2099-02-01 00:00:00.000000')]: eq(same(a,b),False,'Strict typed response mutation rejected') print(json.dumps({'status':'PASS_NEW_FIXTURE_CRUD_NATIVE_AND_FULL_DTO','checks':checks,'cross_full_DTO_pairs':cross,'strict_type_mutations_rejected':7,'lanes':summary},ensure_ascii=False,indent=2))