Coverage for integrations / social / integrity_service.py: 61.2%

1""" 

2HevolveSocial - Integrity Verification Service 

3Challenge-response protocol, impression witnessing, consensus verification, 

4fraud scoring. Central service for all anti-fraud logic. 

5""" 

6import os 

7import secrets 

8import logging 

9import statistics 

10import requests 

11from core.http_pool import pooled_get, pooled_post 

12from datetime import datetime, timedelta 

13from typing import Optional, Dict, List 

14 

15from sqlalchemy import func as sqlfunc 

16from sqlalchemy.orm import Session 

17 

18from .models import ( 

19 PeerNode, AdImpression, HostingReward, 

20 NodeAttestation, IntegrityChallenge, FraudAlert, 

21 User, Post, 

22) 

23 

24logger = logging.getLogger('hevolve_social') 

25 

26# Fraud scoring weights 

27FRAUD_WEIGHTS = { 

28 'hash_mismatch': 30.0, 

29 'challenge_fail': 15.0, 

30 'impression_anomaly': 20.0, 

31 'score_jump': 10.0, 

32 'witness_refusal': 5.0, 

33 'collusion_suspected': 25.0, 

34 'reward_velocity_anomaly': 25.0, 

35 'reward_self_dealing': 35.0, 

36 'spark_gaming': 20.0, 

37 'witness_ring': 30.0, 

38 'temporal_clustering': 20.0, 

39 'seal_tamper': 35.0, 

40 'gradient_magnitude_anomaly': 20.0, 

41 'gradient_direction_flip': 25.0, 

42} 

43 

44IMPRESSION_ANOMALY_STDDEV = 3.0 

45SCORE_JUMP_THRESHOLD_PCT = 200 

46FRAUD_BAN_THRESHOLD = 80.0 

47ATTESTATION_EXPIRY_DAYS = 7 

48MIN_WITNESS_PEERS = 1 

49CHALLENGE_TIMEOUT_SECONDS = 30 

50WITNESS_TIMESTAMP_MAX_AGE = 60 # seconds 

51 

52# ── Fraud Score Decay ── 

53# Every audit round, each node's fraud score decays by this amount. 

54# Good behavior over time earns back trust. But ban records persist. 

55FRAUD_SCORE_DECAY_PER_ROUND = 2.0 # Points per audit round (~5 min) 

56FRAUD_SCORE_DECAY_MIN = 0.0 

57 

58# ── Fail2ban: Progressive Ban Durations ── 

59# Each subsequent ban lasts longer. ban_count tracked on PeerNode. 

60# After max_bans (4+), node must petition for human review. 

61FAIL2BAN_DURATIONS = { 

62 1: timedelta(hours=1), # 1st offense: 1 hour 

63 2: timedelta(hours=24), # 2nd offense: 24 hours 

64 3: timedelta(days=7), # 3rd offense: 1 week 

65} 

66FAIL2BAN_MAX_DURATION = timedelta(days=30) # 4th+ offense: 30 days 

67 

68 

69class IntegrityService: 

70 """Central service for node integrity verification and anti-fraud.""" 

71 

72 # ─── Code Hash Verification ─── 

73 

74 @staticmethod 

75 def verify_code_hash(db: Session, node_id: str, 

76 registry_url: str = None) -> Dict: 

77 """Check if node's code_hash matches expected hash. 

78 Priority: master-signed manifest > registry > local computation.""" 

79 peer = db.query(PeerNode).filter_by(node_id=node_id).first() 

80 if not peer or not peer.code_hash: 

81 return {'verified': False, 'details': 'No code hash available'} 

82 

83 expected = None 

84 

85 # Priority 0: Release hash registry (multi-version support) 

86 try: 

87 from security.release_hash_registry import get_release_hash_registry 

88 registry = get_release_hash_registry() 

89 if registry.is_known_release_hash(peer.code_hash): 

90 peer.integrity_status = 'verified' 

91 peer.last_attestation_at = datetime.utcnow() 

92 return {'verified': True, 

93 'details': 'Code hash in release registry'} 

94 except Exception: 

95 pass 

96 

97 # Primary: check against master-signed manifest 

98 try: 

99 from security.master_key import load_release_manifest, verify_release_manifest 

100 manifest = load_release_manifest() 

101 if manifest and verify_release_manifest(manifest): 

102 expected = manifest.get('code_hash', '') 

103 except Exception: 

104 pass 

105 

106 # Fallback: registry 

107 if not expected and registry_url: 

108 expected = IntegrityService.fetch_expected_hash( 

109 registry_url, peer.code_version or peer.version) 

110 

111 if not expected: 

112 # Last resort: compare against own code hash 

113 try: 

114 from security.node_integrity import compute_code_hash 

115 expected = compute_code_hash() 

116 except Exception: 

117 return {'verified': False, 'details': 'Cannot compute local hash'} 

118 

119 if peer.code_hash == expected: 

120 peer.integrity_status = 'verified' 

121 peer.last_attestation_at = datetime.utcnow() 

122 return {'verified': True, 'details': 'Code hash matches'} 

123 else: 

124 IntegrityService.increase_fraud_score( 

125 db, node_id, FRAUD_WEIGHTS['hash_mismatch'], 

126 f'Code hash mismatch: expected {expected[:16]}..., got {peer.code_hash[:16]}...', 

127 {'expected': expected, 'reported': peer.code_hash}) 

128 return {'verified': False, 'details': 'Code hash mismatch'} 

129 

130 @staticmethod 

131 def fetch_expected_hash(registry_url: str, version: str) -> Optional[str]: 

132 """GET expected code hash from central registry.""" 

133 try: 

134 resp = pooled_get( 

135 f"{registry_url}/api/social/integrity/expected-hash", 

136 params={'version': version}, 

137 timeout=5, 

138 ) 

139 if resp.status_code == 200: 

140 return resp.json().get('code_hash') 

141 except requests.RequestException: 

142 pass 

143 return None 

144 

145 # ─── Challenge-Response Protocol ─── 

146 

147 @staticmethod 

148 def create_challenge(db: Session, challenger_node_id: str, 

149 target_node_id: str, target_url: str, 

150 challenge_type: str = 'agent_count_verify') -> Optional[Dict]: 

151 """Create and send a challenge to a target node.""" 

152 nonce = secrets.token_hex(32) 

153 

154 challenge_data = {'type': challenge_type, 'nonce': nonce} 

155 if challenge_type == 'agent_count_verify': 

156 peer = db.query(PeerNode).filter_by(node_id=target_node_id).first() 

157 if peer: 

158 challenge_data['claimed_agent_count'] = peer.agent_count or 0 

159 elif challenge_type == 'stats_probe': 

160 challenge_data['requested_stats'] = ['agent_count', 'post_count'] 

161 elif challenge_type == 'code_hash_check': 

162 challenge_data['request'] = 'code_hash_and_version' 

163 

164 # Sign the challenge 

165 try: 

166 from security.node_integrity import sign_json_payload, get_public_key_hex 

167 challenge_data['challenger_public_key'] = get_public_key_hex() 

168 challenge_data['signature'] = sign_json_payload(challenge_data) 

169 except Exception: 

170 pass 

171 

172 challenge = IntegrityChallenge( 

173 challenger_node_id=challenger_node_id, 

174 target_node_id=target_node_id, 

175 challenge_type=challenge_type, 

176 challenge_nonce=nonce, 

177 challenge_data=challenge_data, 

178 status='pending', 

179 ) 

180 db.add(challenge) 

181 db.flush() 

182 

183 # Send challenge to target node 

184 try: 

185 resp = pooled_post( 

186 f"{target_url}/api/social/integrity/challenge", 

187 json={'challenge_id': challenge.id, **challenge_data, 

188 'challenger_node_id': challenger_node_id}, 

189 timeout=CHALLENGE_TIMEOUT_SECONDS, 

190 ) 

191 if resp.status_code == 200: 

192 response_data = resp.json() 

193 return IntegrityService.evaluate_challenge_response( 

194 db, challenge.id, 

195 response_data.get('response', {}), 

196 response_data.get('signature', '')) 

197 except requests.RequestException: 

198 challenge.status = 'timeout' 

199 IntegrityService.increase_fraud_score( 

200 db, target_node_id, 5.0, 

201 f'Challenge timeout: {challenge_type}') 

202 

203 peer = db.query(PeerNode).filter_by(node_id=target_node_id).first() 

204 if peer: 

205 peer.last_challenge_at = datetime.utcnow() 

206 return challenge.to_dict() 

207 

208 @staticmethod 

209 def handle_challenge(db: Session, challenge_data: dict) -> Dict: 

210 """Handle an incoming challenge from a peer. Compute response from local data.""" 

211 challenge_type = challenge_data.get('type', '') 

212 nonce = challenge_data.get('nonce', '') 

213 response = {'nonce': nonce, 'timestamp': datetime.utcnow().isoformat()} 

214 

215 if challenge_type == 'agent_count_verify': 

216 actual_count = db.query(sqlfunc.count(User.id)).filter_by( 

217 user_type='agent').scalar() or 0 

218 response['agent_count'] = actual_count 

219 # Include sample agent IDs as proof 

220 agents = db.query(User.id, User.username).filter_by( 

221 user_type='agent').limit(5).all() 

222 response['sample_agents'] = [ 

223 {'id': a.id, 'username': a.username} for a in agents] 

224 

225 elif challenge_type == 'stats_probe': 

226 response['agent_count'] = db.query(sqlfunc.count(User.id)).filter_by( 

227 user_type='agent').scalar() or 0 

228 response['post_count'] = db.query(sqlfunc.count(Post.id)).scalar() or 0 

229 

230 elif challenge_type == 'code_hash_check': 

231 try: 

232 from security.node_integrity import compute_code_hash 

233 from integrations.social.peer_discovery import gossip 

234 response['code_hash'] = compute_code_hash() 

235 response['version'] = gossip.version 

236 except Exception: 

237 response['code_hash'] = 'unavailable' 

238 

239 elif challenge_type == 'guardrail_verify': 

240 try: 

241 from security.hive_guardrails import get_guardrail_hash, compute_guardrail_hash 

242 response['guardrail_hash'] = get_guardrail_hash() 

243 # Recompute live to prove it's not cached/stale 

244 response['guardrail_hash_live'] = compute_guardrail_hash() 

245 except Exception: 

246 response['guardrail_hash'] = 'unavailable' 

247 

248 elif challenge_type == 'impression_audit': 

249 ad_id = challenge_data.get('ad_id') 

250 if ad_id: 

251 hour_ago = datetime.utcnow() - timedelta(hours=1) 

252 count = db.query(sqlfunc.count(AdImpression.id)).filter( 

253 AdImpression.ad_id == ad_id, 

254 AdImpression.created_at >= hour_ago, 

255 ).scalar() or 0 

256 response['impression_count'] = count 

257 

258 # Sign the response 

259 try: 

260 from security.node_integrity import sign_json_payload, get_public_key_hex 

261 response['public_key'] = get_public_key_hex() 

262 sig = sign_json_payload(response) 

263 return {'response': response, 'signature': sig} 

264 except Exception: 

265 return {'response': response, 'signature': ''} 

266 

267 @staticmethod 

268 def evaluate_challenge_response(db: Session, challenge_id: str, 

269 response_data: dict, 

270 response_signature: str) -> Dict: 

271 """Evaluate a challenge response. Verify signature and data consistency.""" 

272 challenge = db.query(IntegrityChallenge).filter_by(id=challenge_id).first() 

273 if not challenge: 

274 return {'passed': False, 'details': 'Challenge not found'} 

275 # Prevent replay: reject already-evaluated challenges 

276 if challenge.status != 'pending': 

277 return {'passed': False, 'details': f'Challenge already processed (status={challenge.status})'} 

278 

279 challenge.response_data = response_data 

280 challenge.response_signature = response_signature 

281 challenge.responded_at = datetime.utcnow() 

282 

283 # Verify nonce 

284 if response_data.get('nonce') != challenge.challenge_nonce: 

285 challenge.status = 'failed' 

286 challenge.result_details = 'Nonce mismatch' 

287 IntegrityService.increase_fraud_score( 

288 db, challenge.target_node_id, FRAUD_WEIGHTS['challenge_fail'], 

289 'Challenge failed: nonce mismatch') 

290 return {'passed': False, 'details': 'Nonce mismatch'} 

291 

292 # Verify signature if public key available 

293 public_key = response_data.get('public_key', '') 

294 if public_key and response_signature: 

295 try: 

296 from security.node_integrity import verify_json_signature 

297 if not verify_json_signature(public_key, response_data, response_signature): 

298 challenge.status = 'failed' 

299 challenge.result_details = 'Invalid signature' 

300 IntegrityService.increase_fraud_score( 

301 db, challenge.target_node_id, FRAUD_WEIGHTS['challenge_fail'], 

302 'Challenge failed: invalid signature') 

303 return {'passed': False, 'details': 'Invalid signature'} 

304 except Exception: 

305 pass 

306 

307 # Evaluate based on challenge type 

308 passed = True 

309 details = 'OK' 

310 

311 if challenge.challenge_type == 'agent_count_verify': 

312 claimed = (challenge.challenge_data or {}).get('claimed_agent_count', 0) 

313 actual = response_data.get('agent_count', 0) 

314 # Allow 10% tolerance 

315 if claimed > 0 and actual < claimed * 0.5: 

316 passed = False 

317 details = f'Agent count mismatch: claimed {claimed}, actual {actual}' 

318 

319 elif challenge.challenge_type == 'stats_probe': 

320 peer = db.query(PeerNode).filter_by( 

321 node_id=challenge.target_node_id).first() 

322 if peer: 

323 reported_agents = response_data.get('agent_count', 0) 

324 if peer.agent_count and reported_agents < (peer.agent_count or 0) * 0.5: 

325 passed = False 

326 details = f'Stats mismatch: DB has {peer.agent_count}, node reports {reported_agents}' 

327 

328 elif challenge.challenge_type == 'code_hash_check': 

329 reported_hash = response_data.get('code_hash', '') 

330 peer = db.query(PeerNode).filter_by( 

331 node_id=challenge.target_node_id).first() 

332 if peer and peer.code_hash and reported_hash != peer.code_hash: 

333 passed = False 

334 details = 'Code hash changed since last exchange' 

335 

336 elif challenge.challenge_type == 'guardrail_verify': 

337 reported_hash = response_data.get('guardrail_hash', '') 

338 reported_live = response_data.get('guardrail_hash_live', '') 

339 try: 

340 from security.hive_guardrails import get_guardrail_hash 

341 expected = get_guardrail_hash() 

342 if reported_hash != expected: 

343 passed = False 

344 details = f'Guardrail hash mismatch: expected {expected[:16]}, got {reported_hash[:16]}' 

345 elif reported_live and reported_live != expected: 

346 passed = False 

347 details = f'Guardrail live recompute mismatch — possible tampering' 

348 elif reported_hash != reported_live: 

349 passed = False 

350 details = 'Cached vs live guardrail hash mismatch — values may have drifted' 

351 except Exception: 

352 pass 

353 

354 challenge.status = 'passed' if passed else 'failed' 

355 challenge.result_details = details 

356 challenge.evaluated_at = datetime.utcnow() 

357 

358 if not passed: 

359 IntegrityService.increase_fraud_score( 

360 db, challenge.target_node_id, FRAUD_WEIGHTS['challenge_fail'], 

361 f'Challenge failed: {details}') 

362 else: 

363 # Successful challenge reduces fraud score slightly 

364 IntegrityService.decrease_fraud_score( 

365 db, challenge.target_node_id, 2.0, 

366 f'Challenge passed: {challenge.challenge_type}') 

367 

368 return {'passed': passed, 'details': details} 

369 

370 # ─── Impression Witnessing ─── 

371 

372 # Rate limit: max witness requests per node per hour 

373 _witness_request_counts = {} # node_id -> (count, window_start) 

374 _WITNESS_MAX_PER_HOUR = 100 

375 

376 @staticmethod 

377 def request_nearest_witness(db: Session, impression_id: str, 

378 ad_id: str, requesting_node_id: str) -> Optional[Dict]: 

379 """Find nearest active non-banned peer and request a witness attestation.""" 

380 # Rate limit witness requests per node 

381 now = datetime.utcnow() 

382 counts = IntegrityService._witness_request_counts 

383 entry = counts.get(requesting_node_id) 

384 if entry: 

385 count, window_start = entry 

386 if (now - window_start).total_seconds() > 3600: 

387 counts[requesting_node_id] = (1, now) 

388 elif count >= IntegrityService._WITNESS_MAX_PER_HOUR: 

389 return None # Rate limited 

390 else: 

391 counts[requesting_node_id] = (count + 1, window_start) 

392 else: 

393 counts[requesting_node_id] = (1, now) 

394 

395 peers = db.query(PeerNode).filter( 

396 PeerNode.status == 'active', 

397 PeerNode.integrity_status != 'banned', 

398 PeerNode.node_id != requesting_node_id, 

399 ).limit(5).all() 

400 

401 for peer in peers: 

402 result = IntegrityService._request_witness_from_peer( 

403 db, impression_id, ad_id, requesting_node_id, peer) 

404 if result: 

405 return result 

406 return None 

407 

408 @staticmethod 

409 def _request_witness_from_peer(db: Session, impression_id: str, 

410 ad_id: str, requesting_node_id: str, 

411 peer: PeerNode) -> Optional[Dict]: 

412 """Request a specific peer to witness an impression.""" 

413 nonce = secrets.token_hex(16) 

414 payload = { 

415 'impression_id': impression_id, 

416 'ad_id': ad_id, 

417 'node_id': requesting_node_id, 

418 'timestamp': datetime.utcnow().isoformat(), 

419 'nonce': nonce, 

420 } 

421 try: 

422 from security.node_integrity import sign_json_payload, get_public_key_hex 

423 payload['public_key'] = get_public_key_hex() 

424 payload['signature'] = sign_json_payload(payload) 

425 except Exception: 

426 pass 

427 

428 try: 

429 resp = pooled_post( 

430 f"{peer.url}/api/social/integrity/witness-impression", 

431 json=payload, 

432 timeout=5, 

433 ) 

434 if resp.status_code == 200: 

435 data = resp.json() 

436 if data.get('witnessed'): 

437 # Store attestation 

438 attestation = NodeAttestation( 

439 attester_node_id=peer.node_id, 

440 subject_node_id=requesting_node_id, 

441 attestation_type='impression_witness', 

442 payload_json={'impression_id': impression_id, 'ad_id': ad_id}, 

443 signature=data.get('signature', ''), 

444 attester_public_key=data.get('public_key', peer.public_key or ''), 

445 expires_at=datetime.utcnow() + timedelta(days=ATTESTATION_EXPIRY_DAYS), 

446 ) 

447 db.add(attestation) 

448 db.flush() 

449 return attestation.to_dict() 

450 except requests.RequestException: 

451 pass 

452 return None 

453 

454 @staticmethod 

455 def handle_witness_request(db: Session, witness_data: dict) -> Dict: 

456 """Handle incoming witness request from a peer node.""" 

457 requesting_node_id = witness_data.get('node_id', '') 

458 timestamp_str = witness_data.get('timestamp', '') 

459 signature = witness_data.get('signature', '') 

460 public_key = witness_data.get('public_key', '') 

461 

462 # Check if requesting node is known and not banned 

463 peer = db.query(PeerNode).filter_by(node_id=requesting_node_id).first() 

464 if peer and peer.integrity_status == 'banned': 

465 return {'witnessed': False, 'reason': 'Requesting node is banned'} 

466 

467 # Check timestamp freshness 

468 if timestamp_str: 

469 try: 

470 ts = datetime.fromisoformat(timestamp_str) 

471 age = (datetime.utcnow() - ts).total_seconds() 

472 if age > WITNESS_TIMESTAMP_MAX_AGE: 

473 return {'witnessed': False, 'reason': 'Stale timestamp'} 

474 except (ValueError, TypeError): 

475 pass 

476 

477 # Verify signature (required — omitting signature is not allowed) 

478 if not signature or not public_key: 

479 return {'witnessed': False, 'reason': 'Missing signature or public key'} 

480 try: 

481 from security.node_integrity import verify_json_signature 

482 if not verify_json_signature(public_key, witness_data, signature): 

483 return {'witnessed': False, 'reason': 'Invalid signature'} 

484 except Exception: 

485 return {'witnessed': False, 'reason': 'Signature verification failed'} 

486 

487 # Co-sign the witness 

488 try: 

489 from security.node_integrity import sign_json_payload, get_public_key_hex 

490 witness_response = { 

491 'witnessed': True, 

492 'impression_id': witness_data.get('impression_id'), 

493 'ad_id': witness_data.get('ad_id'), 

494 'nonce': witness_data.get('nonce'), 

495 'witness_timestamp': datetime.utcnow().isoformat(), 

496 } 

497 witness_response['public_key'] = get_public_key_hex() 

498 witness_response['signature'] = sign_json_payload(witness_response) 

499 return witness_response 

500 except Exception: 

501 return {'witnessed': True, 'signature': '', 'public_key': ''} 

502 

503 @staticmethod 

504 def get_impression_witness_count(db: Session, impression_id: str) -> int: 

505 """Count how many peer witnesses confirmed a given impression.""" 

506 return db.query(sqlfunc.count(NodeAttestation.id)).filter( 

507 NodeAttestation.attestation_type == 'impression_witness', 

508 NodeAttestation.is_valid == True, 

509 NodeAttestation.payload_json.contains(impression_id) if hasattr( 

510 NodeAttestation.payload_json, 'contains') else True, 

511 ).scalar() or 0 

512 

513 # ─── Score Consensus ─── 

514 

515 @staticmethod 

516 def probe_peer_stats(peer_url: str, target_node_id: str) -> Optional[Dict]: 

517 """Query a peer for its view of a target node's stats.""" 

518 try: 

519 resp = pooled_get( 

520 f"{peer_url}/api/social/integrity/peer-stats", 

521 params={'node_id': target_node_id}, 

522 timeout=5, 

523 ) 

524 if resp.status_code == 200: 

525 return resp.json() 

526 except requests.RequestException: 

527 pass 

528 return None 

529 

530 @staticmethod 

531 def run_consensus_check(db: Session, target_node_id: str, 

532 peer_urls: List[str]) -> Dict: 

533 """Query multiple peers for their view of target's stats. Flag disagreements.""" 

534 reports = [] 

535 for url in peer_urls[:5]: 

536 report = IntegrityService.probe_peer_stats(url, target_node_id) 

537 if report: 

538 reports.append(report) 

539 

540 if len(reports) < 2: 

541 return {'consensus': True, 'reports': reports, 'anomalies': [], 

542 'details': 'Not enough peers for consensus'} 

543 

544 agent_counts = [r.get('agent_count', 0) for r in reports] 

545 post_counts = [r.get('post_count', 0) for r in reports] 

546 

547 anomalies = [] 

548 if len(set(agent_counts)) > 1: 

549 spread = max(agent_counts) - min(agent_counts) 

550 mean = statistics.mean(agent_counts) if agent_counts else 0 

551 if mean > 0 and spread / mean > 0.5: 

552 anomalies.append({ 

553 'field': 'agent_count', 

554 'values': agent_counts, 

555 'spread_ratio': round(spread / mean, 2), 

556 }) 

557 

558 consensus = len(anomalies) == 0 

559 if not consensus: 

560 IntegrityService.increase_fraud_score( 

561 db, target_node_id, FRAUD_WEIGHTS['score_jump'] * 0.5, 

562 f'Consensus disagreement: {anomalies}', 

563 {'reports': reports, 'anomalies': anomalies}) 

564 

565 return {'consensus': consensus, 'reports': reports, 'anomalies': anomalies} 

566 

567 # ─── Fraud Detection ─── 

568 

569 @staticmethod 

570 def detect_impression_anomaly(db: Session, node_id: str, 

571 period_hours: int = 24) -> Optional[Dict]: 

572 """Compare node's impression rate to network average. Flag if >3 stddev.""" 

573 cutoff = datetime.utcnow() - timedelta(hours=period_hours) 

574 

575 # Get per-node impression counts 

576 node_counts = db.query( 

577 AdImpression.node_id, 

578 sqlfunc.count(AdImpression.id).label('cnt'), 

579 ).filter( 

580 AdImpression.created_at >= cutoff, 

581 AdImpression.node_id.isnot(None), 

582 ).group_by(AdImpression.node_id).all() 

583 

584 if len(node_counts) < 3: 

585 return None # Not enough data 

586 

587 counts = [nc.cnt for nc in node_counts] 

588 target_count = next((nc.cnt for nc in node_counts if nc.node_id == node_id), 0) 

589 

590 if target_count < 50: 

591 return None # Too low to flag 

592 

593 mean = statistics.mean(counts) 

594 stddev = statistics.stdev(counts) if len(counts) > 1 else 0 

595 

596 if stddev == 0: 

597 return None 

598 

599 z_score = (target_count - mean) / stddev 

600 

601 if z_score > IMPRESSION_ANOMALY_STDDEV: 

602 return IntegrityService._create_fraud_alert( 

603 db, node_id, 'impression_anomaly', 'high', 

604 f'Impression rate anomaly: {target_count} impressions ' 

605 f'(z-score={z_score:.1f}, mean={mean:.1f}, stddev={stddev:.1f})', 

606 FRAUD_WEIGHTS['impression_anomaly'], 

607 {'count': target_count, 'mean': round(mean, 2), 

608 'stddev': round(stddev, 2), 'z_score': round(z_score, 2)}) 

609 

610 return None 

611 

612 @staticmethod 

613 def detect_score_jump(db: Session, node_id: str) -> Optional[Dict]: 

614 """Flag if agent_count or post_count jumped >200% in 24h.""" 

615 peer = db.query(PeerNode).filter_by(node_id=node_id).first() 

616 if not peer: 

617 return None 

618 

619 metadata = peer.metadata_json or {} 

620 prev_agents = metadata.get('_prev_agent_count', 0) 

621 prev_posts = metadata.get('_prev_post_count', 0) 

622 current_agents = peer.agent_count or 0 

623 current_posts = peer.post_count or 0 

624 

625 alert = None 

626 if prev_agents > 5 and current_agents > prev_agents * (1 + SCORE_JUMP_THRESHOLD_PCT / 100): 

627 alert = IntegrityService._create_fraud_alert( 

628 db, node_id, 'score_jump', 'medium', 

629 f'Agent count jumped from {prev_agents} to {current_agents} ' 

630 f'({((current_agents / prev_agents) - 1) * 100:.0f}% increase)', 

631 FRAUD_WEIGHTS['score_jump'], 

632 {'prev': prev_agents, 'current': current_agents, 'field': 'agent_count'}) 

633 

634 if prev_posts > 10 and current_posts > prev_posts * (1 + SCORE_JUMP_THRESHOLD_PCT / 100): 

635 alert = IntegrityService._create_fraud_alert( 

636 db, node_id, 'score_jump', 'medium', 

637 f'Post count jumped from {prev_posts} to {current_posts}', 

638 FRAUD_WEIGHTS['score_jump'], 

639 {'prev': prev_posts, 'current': current_posts, 'field': 'post_count'}) 

640 

641 # Store current values as previous for next check 

642 if not metadata: 

643 metadata = {} 

644 metadata['_prev_agent_count'] = current_agents 

645 metadata['_prev_post_count'] = current_posts 

646 metadata['_last_score_check'] = datetime.utcnow().isoformat() 

647 peer.metadata_json = metadata 

648 

649 return alert 

650 

651 @staticmethod 

652 def detect_collusion(db: Session, node_id: str) -> Optional[Dict]: 

653 """Detect if >80% attestations come from a single peer.""" 

654 cutoff = datetime.utcnow() - timedelta(days=ATTESTATION_EXPIRY_DAYS) 

655 attestations = db.query(NodeAttestation).filter( 

656 NodeAttestation.subject_node_id == node_id, 

657 NodeAttestation.created_at >= cutoff, 

658 NodeAttestation.is_valid == True, 

659 ).all() 

660 

661 if len(attestations) < 5: 

662 return None 

663 

664 attester_counts = {} 

665 for a in attestations: 

666 attester_counts[a.attester_node_id] = attester_counts.get( 

667 a.attester_node_id, 0) + 1 

668 

669 total = sum(attester_counts.values()) 

670 for attester_id, count in attester_counts.items(): 

671 ratio = count / total 

672 if ratio > 0.8: 

673 return IntegrityService._create_fraud_alert( 

674 db, node_id, 'collusion_suspected', 'high', 

675 f'Collusion suspected: {count}/{total} attestations ' 

676 f'({ratio:.0%}) from single peer {attester_id[:8]}', 

677 FRAUD_WEIGHTS['collusion_suspected'], 

678 {'dominant_attester': attester_id, 'ratio': round(ratio, 3), 

679 'count': count, 'total': total}) 

680 

681 return None 

682 

683 @staticmethod 

684 def run_full_audit(db: Session, node_id: str, 

685 registry_url: str = None) -> Dict: 

686 """Run all fraud detection checks on a node, including reward hacking. 

687 

688 Also applies fraud score decay and checks ban expiry — good nodes 

689 recover trust over time, banned nodes serve their sentence then 

690 return to 'suspicious' for re-evaluation. 

691 """ 

692 results = {} 

693 

694 # Decay fraud scores + check ban expiry BEFORE running checks. 

695 # This ensures previously-banned nodes get a fair reassessment. 

696 results['decay'] = IntegrityService.apply_fraud_score_decay(db) 

697 

698 results['code_hash'] = IntegrityService.verify_code_hash( 

699 db, node_id, registry_url) 

700 results['impression_anomaly'] = IntegrityService.detect_impression_anomaly( 

701 db, node_id) 

702 results['score_jump'] = IntegrityService.detect_score_jump(db, node_id) 

703 results['collusion'] = IntegrityService.detect_collusion(db, node_id) 

704 results['audit_dominance'] = IntegrityService.verify_audit_dominance( 

705 db, node_id) 

706 

707 # Reward hacking detection 

708 results['reward_velocity'] = IntegrityService.detect_reward_velocity_anomaly( 

709 db, node_id) 

710 results['reward_self_dealing'] = IntegrityService.detect_reward_self_dealing( 

711 db, node_id) 

712 results['spark_gaming'] = IntegrityService.detect_spark_gaming(db, node_id) 

713 

714 # Impression integrity (collusion + tampering) 

715 results['witness_ring'] = IntegrityService.detect_witness_ring(db, node_id) 

716 results['temporal_clustering'] = IntegrityService.detect_temporal_clustering( 

717 db, node_id) 

718 results['seal_integrity'] = IntegrityService.verify_all_sealed_impressions( 

719 db, node_id, limit=50) 

720 

721 peer = db.query(PeerNode).filter_by(node_id=node_id).first() 

722 results['fraud_score'] = peer.fraud_score if peer else 0.0 

723 results['integrity_status'] = peer.integrity_status if peer else 'unknown' 

724 results['ban_count'] = peer.ban_count if peer else 0 

725 results['ban_until'] = peer.ban_until.isoformat() if peer and peer.ban_until else None 

726 

727 # Auto-isolate if multiple reward hacking signals fire 

728 reward_hack_signals = sum(1 for k in ( 

729 'reward_velocity', 'reward_self_dealing', 'spark_gaming' 

730 ) if results.get(k) is not None) 

731 if reward_hack_signals >= 2: 

732 results['isolation'] = IntegrityService.isolate_reward_hacker( 

733 db, node_id, 

734 f'Multiple reward hacking signals: {reward_hack_signals}/3 triggered', 

735 {'signals': {k: results.get(k) for k in ( 

736 'reward_velocity', 'reward_self_dealing', 'spark_gaming' 

737 ) if results.get(k)}}) 

738 

739 # Auto-isolate impression fraud: witness_ring + temporal_clustering together 

740 impression_fraud_signals = sum(1 for k in ( 

741 'witness_ring', 'temporal_clustering', 

742 ) if results.get(k) is not None) 

743 seal_tampered = (results.get('seal_integrity') or {}).get('tampered', 0) > 0 

744 if impression_fraud_signals >= 2 or (impression_fraud_signals >= 1 and seal_tampered): 

745 results['impression_isolation'] = IntegrityService.isolate_reward_hacker( 

746 db, node_id, 

747 f'Impression fraud: {impression_fraud_signals} signals + ' 

748 f'seal_tampered={seal_tampered}', 

749 {'witness_ring': results.get('witness_ring'), 

750 'temporal_clustering': results.get('temporal_clustering'), 

751 'seal_integrity': results.get('seal_integrity')}) 

752 

753 return results 

754 

755 # ─── Post-Update Peer Witness Verification ─── 

756 

757 @staticmethod 

758 def verify_post_update(db: Session, node_id: str, 

759 expected_version: str = '') -> Dict: 

760 """Verify a node after it reports an upgrade. 

761 

762 Challenges code_hash, verifies guardrail_hash, and records 

763 attestation in NodeAttestation table. Called on next gossip round 

764 after a peer announces a new version. 

765 """ 

766 peer = db.query(PeerNode).filter_by(node_id=node_id).first() 

767 if not peer: 

768 return {'verified': False, 'reason': 'peer not found'}