23046560387

Committed 13 Mar 2026 10:23AM UTC coverage: 15.602% (+5.6%) from 9.997%

Build Type

push

github

Committed by Kimi AI

Commit Message

Merge branch 'feature/test-coverage-workflow'

Coverage Stats

2788 of 17869 relevant lines covered (15.6%)

0.16 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

18.33

/tools/sqlmap_integration.py

"""SQLMap Integration für automatisierte SQL Injection Tests"""

import logging
import subprocess
from typing import Dict, List

logger = logging.getLogger(__name__)


class SQLMapScanner:
    """Wrapper für SQLMap SQL Injection Scanner"""

    def __init__(self, sqlmap_path: str = "sqlmap"):
        self.sqlmap_path = sqlmap_path
        self.check_installation()

    def check_installation(self):
        """Prüft ob sqlmap installiert ist"""
        try:
            result = subprocess.run(
                [self.sqlmap_path, "--version"], capture_output=True, text=True
            )
            logger.info(f"SQLMap verfügbar: {result.stdout.strip()}")
        except FileNotFoundError:
            raise RuntimeError(
                "sqlmap nicht gefunden. Installieren: pip install sqlmap"
            )

    def scan_url(
        self,
        url: str,
        risk: int = 1,
        level: int = 1,
        batch: bool = True,
        dump: bool = False,
    ) -> Dict:
        """
        Scannt URL auf SQL Injection.

        Args:
            url: Ziel-URL mit Parametern
            risk: 1-3 (höher = riskanter)
            level: 1-5 (höher = gründlicher)
            batch: Keine User-Interaktion
            dump: Daten dumpen (nur autorisiert!)
        """
        cmd = [
            self.sqlmap_path,
            "-u",
            url,
            "--risk",
            str(risk),
            "--level",
            str(level),
            "--batch" if batch else "",
            "--dump" if dump else "",
            "--output-dir",
            "/tmp/sqlmap_output",
            "--format",
            "json",
        ]

        # Leere Strings entfernen
        cmd = [c for c in cmd if c]

        try:
            logger.info(f"Starte SQLMap Scan: {url}")
            result = subprocess.run(
                cmd, capture_output=True, text=True, timeout=300
            )

            # Parse Output
            output = {
                "url": url,
                "vulnerable": "sqlmap identified" in result.stdout.lower(),
                "stdout": (
                    result.stdout[-5000:]
                    if len(result.stdout) > 5000
                    else result.stdout
                ),
                "stderr": result.stderr,
            }

            return output

        except subprocess.TimeoutExpired:
            return {"url": url, "error": "Timeout after 300s"}
        except Exception as e:
            return {"url": url, "error": str(e)}

    def scan_form(
        self, url: str, data: str, risk: int = 1, level: int = 1
    ) -> Dict:
        """Scannt POST Form auf SQL Injection"""
        cmd = [
            self.sqlmap_path,
            "-u",
            url,
            "--data",
            data,
            "--risk",
            str(risk),
            "--level",
            str(level),
            "--batch",
        ]

        try:
            result = subprocess.run(
                cmd, capture_output=True, text=True, timeout=300
            )
            return {
                "url": url,
                "vulnerable": "sqlmap identified" in result.stdout.lower(),
                "output": result.stdout[-3000:],
            }
        except Exception as e:
            return {"url": url, "error": str(e)}

    def get_databases(self, url: str) -> List[str]:
        """Listet Datenbanken auf (nur autorisiert!)"""
        cmd = [self.sqlmap_path, "-u", url, "--dbs", "--batch"]

        try:
            result = subprocess.run(
                cmd, capture_output=True, text=True, timeout=300
            )
            # Parse databases from output
            databases = []
            for line in result.stdout.split("\n"):
                if "available databases" in line.lower():
                    # Nächste Zeilen enthalten DB-Namen
                    pass
            return databases
        except Exception as e:
            logger.error(f"Fehler: {e}")
            return []


# LangChain Tool
from langchain_core.tools import tool


@tool
def sqlmap_scan(url: str, risk: int = 1, level: int = 1) -> str:
    """
    Scannt URL auf SQL Injection Vulnerabilities.

    Args:
        url: Ziel-URL (z.B. "http://target.com/page.php?id=1")
        risk: 1-3 (höher = aggressiver)
        level: 1-5 (höher = gründlicher)
    """
    scanner = SQLMapScanner()
    result = scanner.scan_url(url, risk=risk, level=level, dump=False)

    if result.get("vulnerable"):
        return f"SQL Injection gefunden in {url}!"
    elif "error" in result:
        return f"Fehler: {result['error']}"
    else:
        return f"Keine SQL Injection in {url} gefunden."


@tool
def sqlmap_scan_form(url: str, data: str) -> str:
    """Scannt POST Form auf SQL Injection"""
    scanner = SQLMapScanner()
    result = scanner.scan_form(url, data)
    return "Vulnerable" if result.get("vulnerable") else "Not vulnerable"

1	"""SQLMap Integration für automatisierte SQL Injection Tests"""
2
3	import logging	1✔
4	import subprocess	1✔
5	from typing import Dict, List	1✔
6
7	logger = logging.getLogger(__name__)	1✔
8
9
10	class SQLMapScanner:	1✔
11	"""Wrapper für SQLMap SQL Injection Scanner"""
12
13	def __init__(self, sqlmap_path: str = "sqlmap"):	1✔
14	self.sqlmap_path = sqlmap_path	×
15	self.check_installation()	×
16
17	def check_installation(self):	1✔
18	"""Prüft ob sqlmap installiert ist"""
19	try:	×
20	result = subprocess.run(	×
21	[self.sqlmap_path, "--version"], capture_output=True, text=True
22	)
23	logger.info(f"SQLMap verfügbar: {result.stdout.strip()}")	×
24	except FileNotFoundError:	×
25	raise RuntimeError(	×
26	"sqlmap nicht gefunden. Installieren: pip install sqlmap"
27	)
28
29	def scan_url(	1✔
30	self,
31	url: str,
32	risk: int = 1,
33	level: int = 1,
34	batch: bool = True,
35	dump: bool = False,
36	) -> Dict:
37	"""
38	Scannt URL auf SQL Injection.
39
40	Args:
41	url: Ziel-URL mit Parametern
42	risk: 1-3 (höher = riskanter)
43	level: 1-5 (höher = gründlicher)
44	batch: Keine User-Interaktion
45	dump: Daten dumpen (nur autorisiert!)
46	"""
47	cmd = [	×
48	self.sqlmap_path,
49	"-u",
50	url,
51	"--risk",
52	str(risk),
53	"--level",
54	str(level),
55	"--batch" if batch else "",
56	"--dump" if dump else "",
57	"--output-dir",
58	"/tmp/sqlmap_output",
59	"--format",
60	"json",
61	]
62
63	# Leere Strings entfernen
64	cmd = [c for c in cmd if c]	×
65
66	try:	×
67	logger.info(f"Starte SQLMap Scan: {url}")	×
68	result = subprocess.run(	×
69	cmd, capture_output=True, text=True, timeout=300
70	)
71
72	# Parse Output
73	output = {	×
74	"url": url,
75	"vulnerable": "sqlmap identified" in result.stdout.lower(),
76	"stdout": (
77	result.stdout[-5000:]
78	if len(result.stdout) > 5000
79	else result.stdout
80	),
81	"stderr": result.stderr,
82	}
83
84	return output	×
85
86	except subprocess.TimeoutExpired:	×
87	return {"url": url, "error": "Timeout after 300s"}	×
88	except Exception as e:	×
89	return {"url": url, "error": str(e)}	×
90
91	def scan_form(	1✔
92	self, url: str, data: str, risk: int = 1, level: int = 1
93	) -> Dict:
94	"""Scannt POST Form auf SQL Injection"""
95	cmd = [	×
96	self.sqlmap_path,
97	"-u",
98	url,
99	"--data",
100	data,
101	"--risk",
102	str(risk),
103	"--level",
104	str(level),
105	"--batch",
106	]
107
108	try:	×
109	result = subprocess.run(	×
110	cmd, capture_output=True, text=True, timeout=300
111	)
112	return {	×
113	"url": url,
114	"vulnerable": "sqlmap identified" in result.stdout.lower(),
115	"output": result.stdout[-3000:],
116	}
117	except Exception as e:	×
118	return {"url": url, "error": str(e)}	×
119
120	def get_databases(self, url: str) -> List[str]:	1✔
121	"""Listet Datenbanken auf (nur autorisiert!)"""
122	cmd = [self.sqlmap_path, "-u", url, "--dbs", "--batch"]	×
123
124	try:	×
125	result = subprocess.run(	×
126	cmd, capture_output=True, text=True, timeout=300
127	)
128	# Parse databases from output
129	databases = []	×
130	for line in result.stdout.split("\n"):	×
131	if "available databases" in line.lower():	×
132	# Nächste Zeilen enthalten DB-Namen
133	pass	×
134	return databases	×
135	except Exception as e:	×
136	logger.error(f"Fehler: {e}")	×
137	return []	×
138
139
140	# LangChain Tool
141	from langchain_core.tools import tool	1✔
142
143
144	@tool	×
145	def sqlmap_scan(url: str, risk: int = 1, level: int = 1) -> str:	×
146	"""
147	Scannt URL auf SQL Injection Vulnerabilities.
148
149	Args:
150	url: Ziel-URL (z.B. "http://target.com/page.php?id=1")
151	risk: 1-3 (höher = aggressiver)
152	level: 1-5 (höher = gründlicher)
153	"""
154	scanner = SQLMapScanner()	×
155	result = scanner.scan_url(url, risk=risk, level=level, dump=False)	×
156
157	if result.get("vulnerable"):	×
158	return f"SQL Injection gefunden in {url}!"	×
159	elif "error" in result:	×
160	return f"Fehler: {result['error']}"	×
161	else:
162	return f"Keine SQL Injection in {url} gefunden."	×
163
164
165	@tool	×
166	def sqlmap_scan_form(url: str, data: str) -> str:	×
167	"""Scannt POST Form auf SQL Injection"""
168	scanner = SQLMapScanner()	×
169	result = scanner.scan_form(url, data)	×
170	return "Vulnerable" if result.get("vulnerable") else "Not vulnerable"	×

SHAdd0WTAka / Zen-Ai-Pentest / 23046560387

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous