File size: 3,509 Bytes
2a18357 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 | import asyncio
import sys
import os
import json
from datetime import datetime
# Add the project root to sys.path
sys.path.append(os.getcwd())
from app.agents.orchestrator import HoneypotOrchestrator
from app.config import settings
# ANSI Colors for better visibility
class Colors:
HEADER = '\033[95m'
BLUE = '\033[94m'
CYAN = '\033[96m'
GREEN = '\033[92m'
WARNING = '\033[93m'
FAIL = '\033[91m'
ENDC = '\033[0m'
BOLD = '\033[1m'
async def run_test_case(orchestrator, case_name, message):
print(f"\n{Colors.HEADER}{Colors.BOLD}--- TESTING: {case_name} ---{Colors.ENDC}")
print(f"{Colors.BLUE}Input Message:{Colors.ENDC} {message}")
start_time = datetime.now()
try:
result = await orchestrator.process_message(message=message, conversation_id=f"test_{case_name.lower()}")
end_time = datetime.now()
duration = (end_time - start_time).total_seconds()
print(f"{Colors.GREEN}✅ SUCCESS (took {duration:.2f}s){Colors.ENDC}")
print(f"{Colors.CYAN}Detected Scam:{Colors.ENDC} {result.get('scam_type', 'Unknown')}")
print(f"{Colors.CYAN}Risk Score:{Colors.ENDC} {result.get('risk_score', 0):.2f}")
intel = result.get('extracted_intelligence', {})
if intel:
print(f"{Colors.CYAN}Extracted Intel:{Colors.ENDC} {json.dumps(intel, indent=2)}")
persona = result.get('honeypot_response', {}).get('persona', 'Unknown')
response = result.get('honeypot_response', {}).get('message', 'No response generated')
print(f"{Colors.CYAN}Active Persona:{Colors.ENDC} {persona}")
print(f"{Colors.YELLOW}{Colors.BOLD}Honeypot Reply:{Colors.ENDC} {Colors.YELLOW}{response}{Colors.ENDC}")
if result.get('explanation'):
print(f"{Colors.CYAN}Reasoning:{Colors.ENDC} {result['explanation'][0] if isinstance(result['explanation'], list) else result['explanation']}")
except Exception as e:
print(f"{Colors.FAIL}❌ FAILED: {str(e)}{Colors.ENDC}")
async def main():
print(f"{Colors.HEADER}{Colors.BOLD}🛡️ SENTINEL SCAM HONEYPOT - END-TO-END VERIFICATION{Colors.ENDC}")
print("="*60)
# Initialize Orchestrator
orchestrator = HoneypotOrchestrator()
print("Initializing Agents...")
await orchestrator.initialize()
print("All agents ready.\n")
test_cases = [
{
"name": "BANKING_KYC_SCAM",
"message": "Dear customer, your SBI YONO account is blocked today. Please update your KYC immediately at http://sbi-kcy-service.com or visit our nearest branch. Your reference ID is 55421."
},
{
"name": "LOTTERY_PRIZE_SCAM",
"message": "Congratulations!! You have won 25,00,000 RS from KBC Lucky Draw 2025. To claim your prize money, contact KBC Manager Mr. Amit Sharma on WhatsApp +91-9876543210. Processing fee of 15,000 RS is required."
},
{
"name": "JOB_OFFER_SCAM",
"message": "Part-time job offer! Earn 3000-8000 daily by simple task in your mobile. No experience needed. Contact us on WhatsApp for more details or join our group. Register now at http://india-jobs-wfh.org"
}
]
for case in test_cases:
await run_test_case(orchestrator, case["name"], case["message"])
print("-" * 40)
print(f"\n{Colors.GREEN}{Colors.BOLD}VERIFICATION COMPLETE{Colors.ENDC}")
if __name__ == "__main__":
asyncio.run(main())
|