Spaces:
Running
Running
| """Verifiable local identity ledger and durable, fail-closed EVM outbox. | |
| Only receipt commitments are sent to a chain. No simulation exists in this runtime. | |
| Configuration never implies confirmation; imported JSON contains no trusted TX status. | |
| """ | |
| from __future__ import annotations | |
| import copy | |
| import hashlib | |
| import json | |
| import math | |
| import os | |
| import re | |
| from pathlib import Path | |
| import sqlite3 | |
| import threading | |
| import time | |
| from urllib.parse import urlsplit | |
| from uuid import uuid4 | |
| from filelock import FileLock, Timeout as LockTimeout | |
| from sofiel_audit_receipts import (AuditValidationError, append_receipt, canonical_json, | |
| new_ledger, sign_record, validate_ledger, validate_record) | |
| DEFAULTS = {'mode':'local', 'chain_id':None, 'rpc_url':'', 'contract_address':'', | |
| 'data_dir':'', 'confirmations':3, 'max_gas':150000, 'max_fee_per_gas_wei':None, | |
| 'max_gas_fee_wei':None, 'critical_terms':[], 'critical_intents':[], | |
| 'thresholds':{'integrity_score_capitulation':0.6}} | |
| _ENV = {name:'SOFIEL_AUDIT_'+name.upper() for name in DEFAULTS if name not in | |
| {'critical_terms','critical_intents','thresholds'}} | |
| _INTS = {'chain_id','confirmations','max_gas','max_fee_per_gas_wei','max_gas_fee_wei'} | |
| _POLLABLE = ('PENDING','PREPARED','SUBMITTED','MINED','CONFIRMED','BROADCAST_UNKNOWN','NEEDS_REVIEW', | |
| 'LOCAL_SIGNED','AWAITING_CONFIGURATION') | |
| ANCHOR_RUNTIME_KECCAK='0x0f2f69968ceb9a0f09767e197cf813883d1fcd2c0115f9b79994dfe17f646752' | |
| def _require(value, code): | |
| if not value: raise AuditValidationError(code) | |
| class BlockchainAuditor: | |
| use_mock = False | |
| def __init__(self, config_path='critical_intents.json', db_path=None, start_worker=True, | |
| owner=None, settings=None, w3=None): | |
| self.owner=owner; self.scope=uuid4().hex | |
| self._guard=threading.RLock(); self._stop=threading.Event(); self.worker_thread=None | |
| self.w3=w3; self.contract=None; self.wallet_address=None; self.private_key=None | |
| self.config_errors=[]; self.last_error=None; self._validated=False | |
| self._read_settings(config_path, settings) | |
| directory=Path(self._settings.get('data_dir') or Path(config_path).resolve().parent/'audit-data') | |
| self.db_path=str(Path(db_path).resolve() if db_path else directory.resolve()/'audit-outbox-v1.sqlite3') | |
| Path(self.db_path).parent.mkdir(parents=True, exist_ok=True) | |
| self._file_lock=FileLock(self.db_path+'.worker.lock', timeout=1) | |
| self._init_db() | |
| self._load_key() | |
| self.status='DISABLED' if self.config['mode']=='disabled' else 'LOCAL_SIGNED' if self.private_key else 'LOCAL_UNSIGNED' | |
| if self.config['mode']=='live': self.status='AWAITING_CONFIGURATION' | |
| if self.config_errors: self.status='CONFIGURATION_ERROR' | |
| if owner is not None and self.config['mode']!='disabled': self.sync_from_memory() | |
| if start_worker: self.start_worker() | |
| def _read_settings(self,path,settings): | |
| raw={} | |
| try: | |
| if Path(path).is_file(): | |
| _require(Path(path).stat().st_size<=65536,'audit_config_size_limit') | |
| raw=json.loads(Path(path).read_text(encoding='utf-8')) | |
| _require(type(raw) is dict,'invalid_audit_configuration') | |
| if 'private_key' in raw: | |
| raw.pop('private_key') | |
| self.config_errors.append('audit_key_in_public_config') | |
| except Exception: | |
| raw={}; self.config_errors.append('invalid_audit_configuration') | |
| # Explicit settings injection isolates tests and tools from ambient credentials. | |
| if settings is not None: | |
| _require(type(settings) is dict,'invalid_audit_configuration') | |
| raw.update(settings) | |
| else: | |
| for key,env in _ENV.items(): | |
| value=os.getenv(env) | |
| if value is not None: | |
| if key in _INTS: | |
| try: value=int(value) | |
| except ValueError: self.config_errors.append('invalid_'+key);continue | |
| raw[key]=value | |
| raw['private_key']=os.getenv('SOFIEL_AUDIT_PRIVATE_KEY') | |
| aliases={'critical_keywords':'critical_terms','keywords':'critical_terms','intents':'critical_intents'} | |
| for old,new in aliases.items(): | |
| if old in raw: | |
| raw.setdefault(new,raw[old]);raw.pop(old) | |
| # Legacy mock flags can never enable simulated success. | |
| raw.pop('use_mock_blockchain',None) | |
| unknown=set(raw)-set(DEFAULTS)-{'private_key'} | |
| if unknown: self.config_errors.append('unknown_audit_configuration_field') | |
| self._settings={**copy.deepcopy(DEFAULTS),**{k:v for k,v in raw.items() if k in DEFAULTS or k=='private_key'}} | |
| try: | |
| _require(self._settings['mode'] in {'local','live','disabled'},'invalid_audit_mode') | |
| for field in ('critical_terms','critical_intents'): | |
| rows=self._settings[field] | |
| _require(type(rows) is list and len(rows)<=100 and all(type(v) is str and 0<len(v)<=200 for v in rows),'invalid_'+field) | |
| thresholds=self._settings['thresholds'] | |
| _require(type(thresholds) is dict and not set(thresholds)-{'integrity_score_capitulation','volitional_tension_delta'},'invalid_thresholds') | |
| score=thresholds.get('integrity_score_capitulation',.6) | |
| _require(type(score) in (int,float) and math.isfinite(score) and 0<=score<=1,'invalid_thresholds') | |
| for field in _INTS: | |
| value=self._settings[field] | |
| _require(value is None or type(value) is int and value>0,'invalid_'+field) | |
| _require(1<=self._settings['confirmations']<=10000,'invalid_confirmations') | |
| _require(self._settings['max_gas']<=1000000,'invalid_max_gas') | |
| for field in ('rpc_url','contract_address','data_dir'): | |
| _require(type(self._settings[field]) is str and len(self._settings[field])<=4096,'invalid_'+field) | |
| address=self._settings['contract_address'] | |
| _require(not address or re.fullmatch(r'0x[0-9a-fA-F]{40}',address),'invalid_contract_address') | |
| rpc=self._settings['rpc_url'] | |
| if rpc: | |
| parsed=urlsplit(rpc) | |
| _require(parsed.scheme in {'https','http'} and bool(parsed.hostname),'invalid_rpc_url') | |
| except (AuditValidationError,TypeError,KeyError) as exc: | |
| self.config_errors.append(exc.code if isinstance(exc,AuditValidationError) else 'invalid_audit_configuration') | |
| key=self._settings.get('private_key') | |
| self._settings=copy.deepcopy(DEFAULTS) | |
| self._settings['private_key']=key | |
| self.config={k:copy.deepcopy(self._settings[k]) for k in DEFAULTS if k not in {'rpc_url','data_dir'}} | |
| if 'invalid_contract_address' in self.config_errors: self.config['contract_address']='' | |
| def _load_key(self): | |
| key=self._settings.get('private_key') | |
| if not key: return | |
| try: | |
| from eth_account import Account | |
| account=Account.from_key(key) | |
| self.private_key=key;self.wallet_address=account.address | |
| except Exception: self.config_errors.append('invalid_audit_signing_key') | |
| def mode_label(self): | |
| if self.config['mode']=='disabled': return 'DESHABILITADO' | |
| if self.config['mode']=='local': return 'LOCAL_VERIFICABLE' | |
| return 'EVM_CONFIGURADO' if self._validated else 'EVM_PENDIENTE' | |
| def _db(self): | |
| conn=sqlite3.connect(self.db_path, timeout=10) | |
| conn.row_factory=sqlite3.Row | |
| conn.execute('PRAGMA busy_timeout=10000') | |
| return conn | |
| def _init_db(self): | |
| with self._db() as db: | |
| db.executescript(''' | |
| CREATE TABLE IF NOT EXISTS audit_jobs ( | |
| receipt_hash TEXT PRIMARY KEY, record_json TEXT NOT NULL, signer TEXT, | |
| status TEXT NOT NULL, created REAL NOT NULL, chain_id INTEGER, contract_address TEXT, | |
| tx_hash TEXT, raw_tx TEXT, nonce INTEGER, anchor_tx_hash TEXT, | |
| block_number INTEGER, block_hash TEXT, retries INTEGER NOT NULL DEFAULT 0, | |
| next_try REAL NOT NULL DEFAULT 0, error_code TEXT); | |
| CREATE TABLE IF NOT EXISTS audit_scopes ( | |
| scope TEXT NOT NULL, receipt_hash TEXT NOT NULL, PRIMARY KEY(scope,receipt_hash)); | |
| CREATE TABLE IF NOT EXISTS audit_nonces ( | |
| chain_id INTEGER NOT NULL, signer TEXT NOT NULL, nonce INTEGER NOT NULL, | |
| tx_hash TEXT NOT NULL, receipt_hash TEXT NOT NULL UNIQUE, | |
| PRIMARY KEY(chain_id,signer,nonce)); | |
| ''') | |
| def _enqueue(self,record): | |
| record=validate_record(record) | |
| signed=record['signature'] is not None | |
| status=('LOCAL_SIGNED' if signed else 'LOCAL_UNSIGNED') if self.config['mode']=='local' else ('PENDING' if signed else 'LOCAL_UNSIGNED') | |
| with self._db() as db: | |
| db.execute('''INSERT OR IGNORE INTO audit_jobs | |
| (receipt_hash,record_json,signer,status,created) VALUES(?,?,?,?,?)''', | |
| (record['receipt_hash'],canonical_json(record),record['signer'],status,time.time())) | |
| existing=db.execute('SELECT record_json,raw_tx FROM audit_jobs WHERE receipt_hash=?',(record['receipt_hash'],)).fetchone() | |
| prior=validate_record(json.loads(existing['record_json'])) | |
| _require(prior['receipt']==record['receipt'],'audit_queue_hash_conflict') | |
| if prior['signature'] and signed: | |
| _require(prior['signer'].lower()==record['signer'].lower(),'audit_queue_signer_conflict') | |
| if not prior['signature'] and signed and not existing['raw_tx']: | |
| db.execute('UPDATE audit_jobs SET record_json=?,signer=?,status=? WHERE receipt_hash=?', | |
| (canonical_json(record),record['signer'],status,record['receipt_hash'])) | |
| db.execute('INSERT OR IGNORE INTO audit_scopes VALUES(?,?)',(self.scope,record['receipt_hash'])) | |
| def _owner_lock(self): | |
| _require(self.owner is not None and getattr(self.owner,'memory',None) is not None,'audit_owner_missing') | |
| return getattr(self.owner,'state_lock',self._guard) | |
| def sync_from_memory(self): | |
| if self.config['mode']=='disabled': return | |
| with self._owner_lock(): | |
| data=self.owner.memory.data | |
| old=data.get('identity_audit') | |
| ledger=validate_ledger(old) if old is not None else new_ledger() | |
| if self.private_key: | |
| ledger['receipts']=[sign_record(r,self.private_key) if r['signature'] is None else r for r in ledger['receipts']] | |
| if old!=ledger: | |
| data['identity_audit']=ledger | |
| try: | |
| if not self.owner.memory.save(): raise AuditValidationError('audit_memory_save_failed') | |
| except Exception: | |
| if old is None: data.pop('identity_audit',None) | |
| else: data['identity_audit']=old | |
| raise AuditValidationError('audit_memory_save_failed') from None | |
| for record in ledger['receipts']: self._enqueue(record) | |
| def reset_scope(self): | |
| self.scope=uuid4().hex | |
| if self.owner is not None: self.sync_from_memory() | |
| def capture_decision(self,state,rationale,output,event_type='critical_decision',event_id=None): | |
| if self.config['mode']=='disabled': return None | |
| if 'invalid_audit_signing_key' in self.config_errors: raise AuditValidationError('invalid_audit_signing_key') | |
| with self._owner_lock(): | |
| data=self.owner.memory.data;old=data.get('identity_audit') | |
| ledger=validate_ledger(old) if old is not None else new_ledger() | |
| updated,record=append_receipt(ledger,state,rationale,output,event_type,event_id) | |
| if self.private_key: | |
| record=sign_record(record,self.private_key);updated['receipts'][-1]=record | |
| data['identity_audit']=updated | |
| try: | |
| if not self.owner.memory.save(): raise AuditValidationError('audit_memory_save_failed') | |
| except Exception: | |
| if old is None: data.pop('identity_audit',None) | |
| else: data['identity_audit']=old | |
| raise AuditValidationError('audit_memory_save_failed') from None | |
| self._enqueue(record) | |
| return copy.deepcopy(record) | |
| def checkpoint(self,state): | |
| if self.config['mode']=='disabled': return None | |
| identity_fields={'traits','soul_level','era','code_hash','policy_hash'} | |
| wanted={k:v for k,v in state.items() if k in identity_fields} | |
| with self._owner_lock(): | |
| ledger=self.owner.memory.data.get('identity_audit') | |
| if ledger and ledger['receipts']: | |
| previous=ledger['receipts'][-1]['receipt']['state'] | |
| if wanted=={k:v for k,v in previous.items() if k in identity_fields}: return None | |
| return self.capture_decision(state,'','',event_type='identity_checkpoint') | |
| def is_critical_decision(self,entry): | |
| score=entry.get('integrity_score',1) | |
| threshold=self.config['thresholds'].get('integrity_score_capitulation',.6) | |
| if type(score) in (int,float) and math.isfinite(score) and score<threshold: return True | |
| if entry.get('declared_tension') is True: return True | |
| intents=entry.get('intents',entry.get('intent',[])) | |
| if isinstance(intents,str): intents=[intents] | |
| if isinstance(intents,dict): intents=list(intents) | |
| if isinstance(intents,list) and set(self.config['critical_intents']) & {x for x in intents if isinstance(x,str)}: return True | |
| text=(str(entry.get('anchored_reasoning',''))+' '+str(entry.get('final_expression',''))).casefold() | |
| return any(term.casefold() in text for term in self.config['critical_terms']) | |
| def _live_errors(self): | |
| errors=list(self.config_errors) | |
| if self.config['mode']!='live': return ['live_mode_not_enabled'] | |
| for field in ('chain_id','contract_address','data_dir','max_fee_per_gas_wei','max_gas_fee_wei'): | |
| if not self._settings.get(field): errors.append('missing_'+field) | |
| if not self.private_key: errors.append('missing_audit_signing_key') | |
| if self.w3 is None and not self._settings.get('rpc_url'): errors.append('missing_rpc_url') | |
| if self._settings.get('data_dir'): | |
| try: Path(self.db_path).resolve().relative_to(Path(self._settings['data_dir']).resolve()) | |
| except ValueError: errors.append('outbox_outside_data_directory') | |
| return errors | |
| def _connect(self): | |
| from web3 import Web3 | |
| errors=self._live_errors() | |
| if errors: | |
| self.status='AWAITING_CONFIGURATION';self.last_error=errors[0];return False | |
| try: | |
| if self.w3 is None: | |
| self.w3=Web3(Web3.HTTPProvider(self._settings['rpc_url'],request_kwargs={'timeout':10})) | |
| _require(self.w3.is_connected(),'audit_rpc_unavailable') | |
| _require(self.w3.eth.chain_id==self._settings['chain_id'],'audit_chain_id_mismatch') | |
| address=Web3.to_checksum_address(self._settings['contract_address']) | |
| artifact=json.loads((Path(__file__).parent/'contracts/SofielAuditAnchor.json').read_text(encoding='utf-8')) | |
| source=Path(__file__).parent/'contracts/SofielAuditAnchor.sol' | |
| _require(artifact['deployedBytecodeKeccak'].lower()==ANCHOR_RUNTIME_KECCAK | |
| and Web3.to_hex(Web3.keccak(hexstr=artifact['deployedBytecode'])).lower()==ANCHOR_RUNTIME_KECCAK | |
| and hashlib.sha256(source.read_bytes()).hexdigest()==artifact['sourceSHA256'], | |
| 'audit_contract_artifact_mismatch') | |
| code=self.w3.eth.get_code(address) | |
| _require(bool(code),'audit_contract_missing') | |
| _require(Web3.to_hex(Web3.keccak(code)).lower()==artifact['deployedBytecodeKeccak'].lower(),'audit_contract_code_mismatch') | |
| contract=self.w3.eth.contract(address=address,abi=artifact['abi']) | |
| _require(contract.functions.auditVersion().call()==1,'audit_contract_version_mismatch') | |
| _require(contract.functions.auditSigner().call().lower()==self.wallet_address.lower(),'audit_contract_signer_mismatch') | |
| self.contract=contract;self._validated=True;self.status='LIVE_READY';self.last_error=None | |
| return True | |
| except AuditValidationError as exc: self.last_error=exc.code | |
| except Exception: self.last_error='audit_rpc_or_contract_error' | |
| self._validated=False;self.status='CONNECTION_BLOCKED';return False | |
| def _update(self,receipt_hash,**fields): | |
| _require(set(fields)<={'status','chain_id','contract_address','tx_hash','raw_tx','nonce','anchor_tx_hash', | |
| 'block_number','block_hash','retries','next_try','error_code'},'invalid_queue_update') | |
| with self._db() as db: | |
| db.execute('UPDATE audit_jobs SET '+','.join(k+'=?' for k in fields)+' WHERE receipt_hash=?', | |
| (*fields.values(),receipt_hash)) | |
| def _anchor(self,row): | |
| """Validate readback, exact-contract event and canonical block, not merely TX status.""" | |
| from web3 import Web3 | |
| commitment=bytes.fromhex(row['receipt_hash']) | |
| height=int(self.contract.functions.anchors(commitment).call()) | |
| if not height: return None | |
| block=self.w3.eth.get_block(height) | |
| events=self.contract.events.DecisionAnchored().get_logs(from_block=height,to_block=height, | |
| argument_filters={'receiptHash':commitment,'signer':self.wallet_address}) | |
| valid=[e for e in events if e['address'].lower()==self.contract.address.lower() | |
| and bytes(e['args']['receiptHash'])==commitment | |
| and e['args']['signer'].lower()==self.wallet_address.lower() | |
| and bytes(e['blockHash'])==bytes(block['hash'])] | |
| _require(len(valid)==1,'audit_anchor_event_missing') | |
| tx=self.w3.eth.get_transaction_receipt(valid[0]['transactionHash']) | |
| _require(tx['status']==1 and bytes(tx['blockHash'])==bytes(block['hash']),'audit_anchor_receipt_invalid') | |
| confirmations=self.w3.eth.block_number-height+1 | |
| return {'status':'CONFIRMED' if confirmations>=self._settings['confirmations'] else 'MINED', | |
| 'anchor_tx_hash':Web3.to_hex(valid[0]['transactionHash']), 'block_number':height, | |
| 'block_hash':Web3.to_hex(block['hash']),'error_code':None,'next_try':time.time()+2} | |
| def _prepare(self,row): | |
| from eth_account import Account | |
| from web3 import Web3 | |
| account=self.wallet_address | |
| record=validate_record(json.loads(row['record_json'])) | |
| _require(record['signature'] is not None and record['signer'].lower()==account.lower(),'audit_record_signer_mismatch') | |
| price=int(self.w3.eth.gas_price) | |
| _require(price<=self._settings['max_fee_per_gas_wei'],'audit_gas_price_cap') | |
| balance=int(self.w3.eth.get_balance(account)) | |
| _require(balance>0,'audit_insufficient_gas_funds') | |
| fn=self.contract.functions.anchorDecision(bytes.fromhex(row['receipt_hash'])) | |
| estimated=int(fn.estimate_gas({'from':account})) | |
| gas=(estimated*12)//10+1000 | |
| _require(gas<=self._settings['max_gas'] and gas*price<=self._settings['max_gas_fee_wei'],'audit_gas_budget_cap') | |
| _require(balance>=gas*price,'audit_insufficient_gas_funds') | |
| chain=self._settings['chain_id'] | |
| with self._db() as db: | |
| db.execute('BEGIN IMMEDIATE') | |
| reserved=db.execute('SELECT MAX(nonce) FROM audit_nonces WHERE chain_id=? AND signer=?',(chain,account.lower())).fetchone()[0] | |
| nonce=max(int(self.w3.eth.get_transaction_count(account,'pending')),0 if reserved is None else reserved+1) | |
| tx=fn.build_transaction({'from':account,'chainId':chain,'nonce':nonce,'gas':gas,'gasPrice':price,'value':0}) | |
| signed=Account.sign_transaction(tx,self.private_key) | |
| raw=bytes(signed.raw_transaction) | |
| tx_hash=Web3.to_hex(Web3.keccak(raw)) | |
| db.execute('INSERT INTO audit_nonces VALUES(?,?,?,?,?)',(chain,account.lower(),nonce,tx_hash,row['receipt_hash'])) | |
| db.execute('''UPDATE audit_jobs SET status='PREPARED',chain_id=?,contract_address=?,nonce=?, | |
| tx_hash=?,raw_tx=?,error_code=NULL WHERE receipt_hash=?''', | |
| (chain,self.contract.address,nonce,tx_hash,'0x'+raw.hex(),row['receipt_hash'])) | |
| with self._db() as db: | |
| return db.execute('SELECT * FROM audit_jobs WHERE receipt_hash=?',(row['receipt_hash'],)).fetchone() | |
| def _validate_raw_transaction(self,row): | |
| """Recheck persisted legacy EIP-155 bytes before rebroadcast.""" | |
| from eth_account import Account | |
| from eth_account._utils.legacy_transactions import Transaction | |
| from web3 import Web3 | |
| try: | |
| raw=bytes.fromhex(row['raw_tx'][2:]) | |
| _require(0<len(raw)<=4096,'audit_raw_transaction_invalid') | |
| tx=Transaction.from_bytes(raw).as_dict() | |
| _require(Web3.to_hex(Web3.keccak(raw)).lower()==row['tx_hash'].lower(),'audit_raw_transaction_mismatch') | |
| _require(Account.recover_transaction(raw).lower()==self.wallet_address.lower(),'audit_raw_signer_mismatch') | |
| _require(tx['v']>=35 and (tx['v']-35)//2==self._settings['chain_id'],'audit_raw_chain_mismatch') | |
| expected=self.contract.functions.anchorDecision(bytes.fromhex(row['receipt_hash']))._encode_transaction_data() | |
| _require(tx['nonce']==row['nonce'] and bytes(tx['to'])==bytes.fromhex(self.contract.address[2:]) | |
| and tx['value']==0 and bytes(tx['data'])==bytes.fromhex(expected[2:]),'audit_raw_target_mismatch') | |
| _require(0<tx['gas']<=self._settings['max_gas'] and 0<=tx['gasPrice']<=self._settings['max_fee_per_gas_wei'] | |
| and tx['gas']*tx['gasPrice']<=self._settings['max_gas_fee_wei'],'audit_raw_fee_cap') | |
| with self._db() as db: | |
| reserved=db.execute('SELECT tx_hash,receipt_hash FROM audit_nonces WHERE chain_id=? AND signer=? AND nonce=?', | |
| (self._settings['chain_id'],self.wallet_address.lower(),row['nonce'])).fetchone() | |
| _require(reserved is not None and reserved['tx_hash']==row['tx_hash'] | |
| and reserved['receipt_hash']==row['receipt_hash'],'audit_nonce_reservation_mismatch') | |
| return raw | |
| except AuditValidationError: raise | |
| except Exception: | |
| raise AuditValidationError('audit_raw_transaction_invalid') from None | |
| def _reconcile(self,row): | |
| from web3 import Web3 | |
| from web3.exceptions import TransactionNotFound | |
| h=row['receipt_hash'] | |
| if row['chain_id'] is not None: | |
| _require(row['chain_id']==self._settings['chain_id'] and row['contract_address'].lower()==self.contract.address.lower(), | |
| 'audit_destination_changed') | |
| record=validate_record(json.loads(row['record_json'])) | |
| _require(record['signer'] and record['signer'].lower()==self.wallet_address.lower(),'audit_record_signer_mismatch') | |
| evidence=self._anchor(row) | |
| if evidence: | |
| self._update(h,chain_id=self._settings['chain_id'],contract_address=self.contract.address,**evidence) | |
| return True | |
| if row['status'] in {'CONFIRMED','MINED'}: | |
| self._update(h,status='SUBMITTED',block_number=None,block_hash=None,anchor_tx_hash=None,error_code='audit_reorg_detected') | |
| if not row['raw_tx']: row=self._prepare(row) | |
| else: | |
| self._validate_raw_transaction(row) | |
| try: | |
| tx=self.w3.eth.get_transaction_receipt(row['tx_hash']) | |
| if tx['status']==0: | |
| self._update(h,status='FAILED',error_code='audit_transaction_reverted');return False | |
| self._update(h,status='NEEDS_REVIEW',error_code='audit_anchor_event_missing',next_try=time.time()+10) | |
| return False | |
| except TransactionNotFound: pass | |
| # A lost response never creates another nonce/TX: resend exactly the persisted bytes. | |
| if row['retries']>=5: | |
| self._update(h,status='NEEDS_REVIEW',error_code='audit_broadcast_retry_limit',next_try=time.time()+30);return False | |
| try: | |
| if self._stop.is_set(): return False | |
| sent=self.w3.eth.send_raw_transaction(self._validate_raw_transaction(row)) | |
| _require(Web3.to_hex(sent).lower()==row['tx_hash'].lower(),'audit_broadcast_hash_mismatch') | |
| self._update(h,status='SUBMITTED',retries=row['retries']+1,error_code=None,next_try=time.time()+2) | |
| evidence=self._anchor(row) | |
| if evidence: self._update(h,**evidence) | |
| return True | |
| except AuditValidationError: raise | |
| except Exception: | |
| self._update(h,status='BROADCAST_UNKNOWN',retries=row['retries']+1, | |
| error_code='audit_broadcast_unresolved',next_try=time.time()+min(60,2**(row['retries']+1))) | |
| return False | |
| def process_once(self): | |
| """One bounded worker step, also usable with a local provider in tests.""" | |
| if self.config['mode']!='live' or self._stop.is_set(): return False | |
| try: | |
| with self._file_lock: | |
| if not self._connect() or self._stop.is_set(): return False | |
| with self._db() as db: | |
| row=db.execute('''SELECT * FROM audit_jobs WHERE status IN ('''+','.join('?' for _ in _POLLABLE)+''') | |
| AND signer=? COLLATE NOCASE AND next_try<=? ORDER BY next_try,created LIMIT 1''', | |
| (*_POLLABLE,self.wallet_address,time.time())).fetchone() | |
| if row is None: return False | |
| try: return self._reconcile(row) | |
| except AuditValidationError as exc: | |
| self.last_error=exc.code | |
| self._update(row['receipt_hash'],status='NEEDS_REVIEW',error_code=exc.code,next_try=time.time()+10) | |
| return False | |
| except Exception: | |
| self.last_error='audit_reconciliation_error' | |
| self._update(row['receipt_hash'],error_code=self.last_error,next_try=time.time()+10) | |
| return False | |
| except LockTimeout: return False | |
| def _worker(self): | |
| while not self._stop.is_set(): | |
| try: self.process_once() | |
| except Exception: self.last_error='audit_worker_error' | |
| self._stop.wait(2) | |
| def start_worker(self): | |
| if self.config['mode']!='live' or self._live_errors(): return False | |
| with self._guard: | |
| if self.worker_thread and self.worker_thread.is_alive(): return True | |
| self._stop.clear() | |
| self.worker_thread=threading.Thread(target=self._worker,name='sofiel-audit-outbox',daemon=True) | |
| self.worker_thread.start() | |
| return True | |
| def stop(self): | |
| self._stop.set() | |
| thread=self.worker_thread | |
| if thread and thread is not threading.current_thread(): thread.join(timeout=15) | |
| if thread and thread.is_alive(): | |
| self.last_error='audit_worker_stop_timeout';return False | |
| return True | |
| def get_queue_stats(self): | |
| with self._db() as db: | |
| rows=db.execute('''SELECT j.status,COUNT(*) AS count FROM audit_jobs j JOIN audit_scopes s | |
| ON j.receipt_hash=s.receipt_hash WHERE s.scope=? GROUP BY j.status''',(self.scope,)).fetchall() | |
| recent=db.execute('''SELECT j.receipt_hash,j.status,j.tx_hash,j.anchor_tx_hash,j.block_number,j.error_code | |
| FROM audit_jobs j JOIN audit_scopes s ON j.receipt_hash=s.receipt_hash | |
| WHERE s.scope=? ORDER BY j.created DESC LIMIT 5''',(self.scope,)).fetchall() | |
| return {'counts':{r['status']:r['count'] for r in rows},'recent':[dict(r) for r in recent], | |
| 'mode':self.mode_label,'status':self.status,'configuration_errors':list(dict.fromkeys(self.config_errors+self._live_errors())) if self.config['mode']=='live' else list(self.config_errors), | |
| 'last_error':self.last_error,'wallet':self.wallet_address,'chain_id':self._settings['chain_id'], | |
| 'contract':self._settings['contract_address'],'confirmations_required':self._settings['confirmations']} | |
| def report(self): | |
| stats=self.get_queue_stats() | |
| lines=['### AUDITORÍA DE IDENTIDAD — recibos verificables', | |
| '**Modo:** '+stats['mode'], '**Estado:** '+stats['status'], | |
| '**Identidad:** '+str((getattr(getattr(self.owner,'memory',None),'data',{}) or {}).get('identity_audit',{}).get('identity_id','sin recibos')), | |
| '**Firmante:** '+str(stats['wallet'] or 'sin clave de auditoría'), | |
| '**Cadena configurada:** '+str(stats['chain_id'] or 'pendiente'), | |
| '**Contrato:** '+str(stats['contract'] or 'pendiente'), | |
| '**Confirmaciones requeridas:** '+str(stats['confirmations_required'])] | |
| if stats['configuration_errors']: lines.append('**Pendientes:** '+', '.join(stats['configuration_errors'])) | |
| if stats['last_error']: lines.append('**Diagnóstico:** '+stats['last_error']) | |
| lines.append('**Recibos de esta sesión:** '+(canonical_json(stats['counts']) if stats['counts'] else 'sin recibos')) | |
| for row in stats['recent']: | |
| lines.append('- `'+row['receipt_hash']+'` — '+row['status']+ | |
| (' — TX `'+str(row['anchor_tx_hash'] or row['tx_hash'])+'`' if row['tx_hash'] or row['anchor_tx_hash'] else '')) | |
| lines.append('Solo CONFIRMED identifica un anclaje validado con las confirmaciones configuradas. ' | |
| 'Los recibos acreditan compromisos y procedencia; no prueban consciencia ni veracidad del modelo.') | |
| lines.append('LOCAL_UNSIGNED conserva hashes y encadenado sin autenticar un firmante. ' | |
| 'La cadena no está activa hasta configurar y verificar la red y el contrato. ' | |
| 'Los estados confirmados reflejan la última comprobación del worker contra el RPC configurado.') | |
| return '\n\n'.join(lines) | |