Spaces:
Running
Running
| """Protected DB2DB shared-cell rendezvous. Operator auth is the existing SMESH key.""" | |
| from __future__ import annotations | |
| import base64,hashlib,hmac,json,secrets,sqlite3,tempfile,time,zipfile | |
| from pathlib import Path | |
| from threading import RLock | |
| _LOCK=RLock();_CELLS={};ROOT=Path(tempfile.gettempdir())/'signalmesh-db2db-cells';ROOT.mkdir(exist_ok=True) | |
| def _sha(p):return hashlib.sha256(Path(p).read_bytes()).hexdigest() | |
| def create_cell(ttl=3600,band=None,site_fingerprint=None): | |
| sid=secrets.token_hex(4).upper();sid=f'{sid[:4]}-{sid[4:]}';key=secrets.token_urlsafe(32);now=int(time.time());band=(str(band).lower() if band else None) | |
| if band not in (None,'source','target'):raise ValueError('band must be source or target') | |
| with _LOCK:_CELLS[sid]={'id':sid,'session_key':key,'created_at':now,'expires_at':now+ttl,'requested_band':band,'requested_site_fingerprint':site_fingerprint,'bands':{'source':None,'target':None},'connections':{'source':None,'target':None},'handshake_nonces':set(),'map':None,'verification':None,'allow_apply':False} | |
| return {'coordinate':sid,'expires_at':now+ttl,'bands':['source','target'],'requested_band':band} | |
| def assign_band(sid,band,site_fingerprint=None): | |
| band=str(band).lower() | |
| if band not in ('source','target'):raise ValueError('band must be source or target') | |
| with _LOCK: | |
| c=_CELLS.get(sid) | |
| if not c:raise KeyError('unknown migration coordinate') | |
| c['requested_band']=band;c['requested_site_fingerprint']=site_fingerprint;c['connections'][band]=None | |
| return status(sid) | |
| def discover(site_fingerprint=None): | |
| now=time.time() | |
| with _LOCK: | |
| candidates=[c for c in _CELLS.values() if now<=c['expires_at'] and c.get('requested_band') and not c['bands'][c['requested_band']] and (not c.get('requested_site_fingerprint') or c.get('requested_site_fingerprint')==site_fingerprint)] | |
| if not candidates:return None | |
| c=max(candidates,key=lambda x:x['created_at']);band=c['requested_band'] | |
| return {'pending':True,'coordinate':c['id'],'band':band,'session_key':c['session_key'],'relay_url':'/api/editor/api/migration/beacon','handshake_url':'/api/editor/api/migration/beacon/handshake','expires_at':c['expires_at'],'allow_apply':bool(c.get('allow_apply') and band=='target')} | |
| def handshake(env): | |
| sid=str(env.get('coordinate',''));band=str(env.get('band','')).lower();fp=str(env.get('site_fingerprint',''));nonce=str(env.get('nonce',''));protocol=str(env.get('protocol','signalmesh.db2db.v2')) | |
| try:ts=int(env.get('timestamp',0)) | |
| except Exception:raise PermissionError('invalid handshake timestamp') | |
| if protocol!='signalmesh.db2db.v2':raise PermissionError('unsupported DB2DB protocol') | |
| if band not in ('source','target') or not sid or not fp or not nonce:raise PermissionError('incomplete DB2DB handshake') | |
| now=int(time.time()) | |
| if abs(now-ts)>120:raise TimeoutError('DB2DB handshake timestamp outside allowed window') | |
| with _LOCK: | |
| c=_CELLS.get(sid) | |
| if not c:raise KeyError('unknown migration coordinate') | |
| if now>c['expires_at']:raise TimeoutError('migration coordinate expired') | |
| if c.get('requested_band') and c['requested_band']!=band:raise PermissionError('wrong reality band for this session') | |
| if c.get('requested_site_fingerprint') and c['requested_site_fingerprint']!=fp:raise PermissionError('site fingerprint does not match listener') | |
| if nonce in c['handshake_nonces']:raise PermissionError('replayed DB2DB handshake') | |
| signed='|'.join([sid,band,fp,nonce,str(ts)]);want=hmac.new(c['session_key'].encode(),signed.encode(),hashlib.sha256).hexdigest() | |
| if not hmac.compare_digest(want,str(env.get('signature',''))):raise PermissionError('invalid handshake signature') | |
| c['handshake_nonces'].add(nonce);c['connections'][band]={'site_fingerprint':fp,'plugin_version':env.get('plugin_version'),'joomla_version':env.get('joomla_version'),'connected_at':now,'protocol':protocol} | |
| return {'connected':True,'coordinate':sid,'band':band,'site_fingerprint':fp,'connected_at':now,'protocol':protocol} | |
| def authorize_apply(sid,allowed=True): | |
| with _LOCK: | |
| c=_CELLS.get(sid) | |
| if not c:raise KeyError('unknown migration coordinate') | |
| c['allow_apply']=bool(allowed) | |
| return status(sid) | |
| def status(sid): | |
| with _LOCK:c=_CELLS.get(sid) | |
| if not c:return None | |
| return {'coordinate':sid,'expires_at':c['expires_at'],'requested_band':c.get('requested_band'),'connections':{k:(None if not v else dict(v)) for k,v in c.get('connections',{}).items()},'bands':{k:(None if not v else {x:v[x] for x in ('site_fingerprint','joomla_version','artifact_sha256','received_at')}) for k,v in c['bands'].items()},'ready':all(c['bands'].values()),'map':c.get('map'),'verification':c.get('verification'),'allow_apply':bool(c.get('allow_apply'))} | |
| def receive(env,replace=False): | |
| sid=str(env.get('session_id',''));band=str(env.get('band','')).lower() | |
| if band not in ('source','target'):raise ValueError('band must be source or target') | |
| with _LOCK: | |
| c=_CELLS.get(sid) | |
| if not c:raise KeyError('unknown migration coordinate') | |
| if time.time()>c['expires_at']:raise TimeoutError('migration coordinate expired') | |
| if c.get('requested_band') and c['requested_band']!=band:raise PermissionError('wrong reality band for this session') | |
| connection=c.get('connections',{}).get(band) | |
| if not connection:raise PermissionError('DB2DB handshake required before artifact transmission') | |
| if connection.get('site_fingerprint')!=str(env.get('site_fingerprint','')):raise PermissionError('artifact site does not match connected beacon') | |
| if c['bands'][band] and not replace:raise FileExistsError(f'{band} band is locked') | |
| secret=c['session_key'] | |
| signed='|'.join([sid,band,str(env.get('artifact_sha256','')),str(env.get('manifest_sha256',''))]);want=hmac.new(secret.encode(),signed.encode(),hashlib.sha256).hexdigest() | |
| if not hmac.compare_digest(want,str(env.get('signature',''))):raise PermissionError('invalid envelope signature') | |
| raw=base64.b64decode(env['artifact']);nonce=base64.b64decode(env['nonce']);key=hashlib.sha256(secret.encode()).digest();cipher=env.get('cipher') | |
| if cipher=='aes-256-gcm': | |
| from cryptography.hazmat.primitives.ciphers.aead import AESGCM;raw=AESGCM(key).decrypt(nonce,raw+base64.b64decode(env['tag']),None) | |
| elif cipher=='xsalsa20-poly1305': | |
| from nacl.secret import SecretBox;raw=SecretBox(key).decrypt(raw,nonce) | |
| else:raise ValueError('unsupported cipher') | |
| got=hashlib.sha256(raw).hexdigest() | |
| if not hmac.compare_digest(got,str(env.get('artifact_sha256',''))):raise ValueError('artifact hash mismatch') | |
| d=ROOT/sid;d.mkdir(exist_ok=True);p=d/f'{band}.db';p.write_bytes(raw);rec={'site_fingerprint':env.get('site_fingerprint'),'joomla_version':env.get('joomla_version'),'artifact_sha256':got,'manifest':env.get('manifest',{}),'received_at':int(time.time()),'path':str(p)} | |
| with _LOCK:_CELLS[sid]['bands'][band]=rec;_CELLS[sid]['requested_band']=('target' if band=='source' and not _CELLS[sid]['bands']['target'] else None);_CELLS[sid]['requested_site_fingerprint']=None;_CELLS[sid]['map']=None;_CELLS[sid]['verification']=None | |
| return status(sid) | |
| def _read_portable(path): | |
| db=sqlite3.connect(path);db.row_factory=sqlite3.Row;tables={r['name']:{'columns':json.loads(r['columns_json'] or '{}'),'rows':[]} for r in db.execute('select name,columns_json from migration_tables')} | |
| for r in db.execute('select table_name,row_no,row_json from migration_rows order by table_name,row_no'): | |
| if r['table_name'] in tables:tables[r['table_name']]['rows'].append(json.loads(r['row_json'])) | |
| meta={r['k']:r['v'] for r in db.execute('select k,v from migration_meta')};db.close();return meta,tables | |
| def _identity(row,columns): | |
| for k in ('id','extension_id','asset_id','user_id','catid','menuid','field_id','group_id'): | |
| if k in row and row[k] not in (None,''):return(k,str(row[k])) | |
| for k in ('alias','element','name','title','username','email'): | |
| if k in row and row[k] not in (None,''):return(k,str(row[k])) | |
| return('hash',hashlib.sha256(json.dumps(row,sort_keys=True,default=str).encode()).hexdigest()) | |
| def build_map(sid): | |
| with _LOCK:c=_CELLS.get(sid) | |
| if not c or not all(c['bands'].values()):raise ValueError('source and target are required') | |
| d=ROOT/sid;sm,source=_read_portable(d/'source.db');tm,target=_read_portable(d/'target.db');plan=[];blockers=[];stats={'preserve_target':0,'insert_source':0,'merge_source':0,'source_only_tables':0,'target_only_tables':0} | |
| for name in sorted(set(source)|set(target)): | |
| if name not in source:stats['target_only_tables']+=1;plan.append({'table':name,'action':'preserve-target-table'});continue | |
| if name not in target:stats['source_only_tables']+=1;blockers.append({'table':name,'reason':'table absent from target; extension/schema must exist before migration'});continue | |
| sc=set(source[name]['columns']);tc=set(target[name]['columns']) | |
| if not sc&tc:blockers.append({'table':name,'reason':'no compatible columns'});continue | |
| target_index={_identity(r,tc):r for r in target[name]['rows']} | |
| for row in source[name]['rows']: | |
| ident=_identity(row,sc);clean={k:v for k,v in row.items() if k in tc} | |
| if ident in target_index: | |
| patch={k:v for k,v in clean.items() if target_index[ident].get(k) in (None,'') and v not in (None,'')} | |
| if patch:plan.append({'table':name,'action':'hydrate','identity':list(ident),'values':patch});stats['merge_source']+=1 | |
| else:stats['preserve_target']+=1 | |
| else:plan.append({'table':name,'action':'insert','identity':list(ident),'values':clean});stats['insert_source']+=1 | |
| result={'schema':'signalmesh.db2db.map.v1','source':sm,'target':tm,'stats':stats,'blockers':blockers,'operations':plan,'ready':not blockers,'created_at':int(time.time())};(d/'migration-map.json').write_text(json.dumps(result,indent=2,default=str)) | |
| with _LOCK:_CELLS[sid]['map']={'ready':result['ready'],'stats':stats,'blockers':blockers,'operation_count':len(plan)} | |
| return result | |
| def construct_merged(sid): | |
| plan=build_map(sid);d=ROOT/sid | |
| if plan['blockers']:raise ValueError('migration map has blockers; install/map required target schemas first') | |
| merged=d/'merged.db';merged.write_bytes((d/'target.db').read_bytes());db=sqlite3.connect(merged) | |
| for op in plan['operations']: | |
| if op['action'] not in ('insert','hydrate'):continue | |
| table=op['table'];ident_key,ident_val=op['identity'];values=op['values'] | |
| if op['action']=='insert':row_no=db.execute('select coalesce(max(row_no),-1)+1 from migration_rows where table_name=?',(table,)).fetchone()[0];db.execute('insert into migration_rows(table_name,row_no,row_json) values(?,?,?)',(table,row_no,json.dumps(values,default=str))) | |
| else: | |
| for rn,rj in db.execute('select row_no,row_json from migration_rows where table_name=?',(table,)).fetchall(): | |
| row=json.loads(rj) | |
| if _identity(row,set(row))==(ident_key,ident_val):row.update(values);db.execute('update migration_rows set row_json=? where table_name=? and row_no=?',(json.dumps(row,default=str),table,rn));break | |
| db.commit();db.close();return str(merged),plan | |
| def verify(sid,actual_path=None): | |
| d=ROOT/sid;expected=d/'merged.db';actual=Path(actual_path) if actual_path else d/'target.db' | |
| if not expected.exists():raise ValueError('merged future has not been constructed') | |
| _,e=_read_portable(expected);_,a=_read_portable(actual);diffs=[] | |
| for name in sorted(set(e)|set(a)): | |
| er=e.get(name,{}).get('rows',[]);ar=a.get(name,{}).get('rows',[]) | |
| if len(er)!=len(ar):diffs.append({'table':name,'expected_rows':len(er),'actual_rows':len(ar)}) | |
| result={'schema':'signalmesh.db2db.verify.v1','verified':not diffs,'differences':diffs,'expected_sha256':_sha(expected),'actual_sha256':_sha(actual),'checked_at':int(time.time())};(d/'verification.json').write_text(json.dumps(result,indent=2)) | |
| with _LOCK:_CELLS[sid]['verification']=result | |
| return result | |
| def build_bundle(sid): | |
| d=ROOT/sid;merged,plan=construct_merged(sid);source=d/'source.db';target=d/'target.db';manifest={'schema':'signalmesh.db2db.bundle.v2','coordinate':sid,'source_sha256':_sha(source),'target_sha256':_sha(target),'merged_sha256':_sha(merged),'map_ready':plan['ready'],'operations':len(plan['operations']),'created_at':int(time.time())};(d/'manifest.json').write_text(json.dumps(manifest,indent=2));out=d/'migration-bundle.zip' | |
| with zipfile.ZipFile(out,'w',zipfile.ZIP_DEFLATED) as z:z.write(source,'01-source.db');z.write(target,'02-target.db');z.write(merged,'03-merged.db');z.write(d/'migration-map.json','migration-map.json');z.write(d/'manifest.json','manifest.json') | |
| return str(out),manifest |