Вопрос или проблема
Я использую библиотеку requests в Python для выполнения HTTP-запросов и сталкиваюсь с ошибками соединения, такими как превышение максимального числа попыток и не удалось установить новое соединение. Эти ошибки выводятся в консоль, несмотря на мои усилия их подавить.
Я попробовал несколько подходов, таких как подавление логов от urllib3 и requests с помощью модуля logging, игнорирование предупреждений с помощью модуля warnings, оборачивание запросов в блоки try/except для тихой обработки исключений.
Тем не менее, ошибки все равно появляются в консоли, как это:
Ошибка при получении https://TEST.com/sites/default/modules/advanced_comment_trigger/LICENSE.txt с методом GET: HTTPSConnectionPool(host="TEST.com", port=443): Превышено максимальное число попыток с url: /sites/default/modules/advanced_comment_trigger/LICENSE.txt (вызвано NewConnectionError('<urllib3.connection.HTTPSConnection object at 0x102bd1a00>: Не удалось установить новое соединение: [Errno 8] имя узла или имя службы не указаны, или неизвестно'))
Мой код:
import os
from concurrent.futures import ThreadPoolExecutor, as_completed
from bs4 import BeautifulSoup
import requests
import logging
import warnings
import threading
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
warnings.filterwarnings("ignore", category=UserWarning, module="urllib3")
logging.getLogger("urllib3").setLevel(logging.CRITICAL)
logging.getLogger("requests").setLevel(logging.CRITICAL)
DEBUG = False
MAX_WORKERS = 20
MIN_WORKERS = 1
WORKER_DECREMENT = 5
lock = threading.Lock()
def create_retry_session():
session = requests.Session()
retries = Retry(total=3, backoff_factor=0.1, status_forcelist=[429, 500, 502, 503, 504])
adapter = HTTPAdapter(max_retries=retries)
session.mount('http://', adapter)
session.mount('https://', adapter)
return session
def load_drupal_modules(db_file="databases/drupal"):
drupal_modules = []
try:
with open(db_file, 'r') as f:
for line in f:
parts = line.strip().split(',')
if len(parts) == 2:
drupal_modules.append({
'test_id': parts[0].strip().strip('"'),
'module': parts[1].strip().strip('"')
})
except FileNotFoundError:
return drupal_modules
return drupal_modules
def is_drupal_site(url, fetch_url, user_agent):
response = fetch_url(url, method="GET", headers={"User-Agent": user_agent})
if response:
if 'X-Generator' in response.headers and 'Drupal' in response.headers['X-Generator']:
return True
soup = BeautifulSoup(response.text, 'html.parser')
meta_generator = soup.find('meta', attrs={'name': 'Generator'})
if meta_generator and 'Drupal' in meta_generator.get('content', ''):
return True
drupal_paths = ['/user', '/node', '/CHANGELOG.txt', '/sites/default/files/']
for path in drupal_paths:
check_url = url + path
check_response = fetch_url(check_url, method="HEAD", headers={"User-Agent": user_agent})
if check_response and check_response.status_code == 200:
return True
return False
def check_module(url, fetch_url, user_agent, module, module_path="/sites/default/modules"):
module_url = f"{module_path}/{module['module']}/LICENSE.txt"
full_url = url.lower() + module_url
if DEBUG:
print(f"+ Testing URL: {full_url}")
try:
response = fetch_url(full_url, method="GET", headers={"User-Agent": user_agent})
if response and response.status_code == 200 and 'LICENSE' in response.text:
return f"+ {full_url}: Модуль Drupal {module['module']} найден.\n test ID: {module['test_id']}"
except requests.exceptions.RequestException:
reduce_thread_count()
return None
def reduce_thread_count():
global MAX_WORKERS
with lock:
if MAX_WORKERS > MIN_WORKERS:
MAX_WORKERS = max(MIN_WORKERS, MAX_WORKERS - WORKER_DECREMENT) # Уменьшить количество потоков
if DEBUG:
print(f"- Уменьшение количества потоков до {MAX_WORKERS}")
def enumerate_drupal_modules(url, fetch_url, user_agent, module_path="/sites/default/modules"):
drupal_modules = load_drupal_modules()
if not drupal_modules:
return
found_vulnerability = False
completed_tests = []
with ThreadPoolExecutor(max_workers=MAX_WORKERS) as executor:
futures = [executor.submit(check_module, url, fetch_url, user_agent, module, module_path) for module in drupal_modules]
for future in as_completed(futures):
result = future.result()
completed_tests.append(result)
if result:
print(result)
found_vulnerability = True
if not found_vulnerability:
print("+ Тест завершен. Ни один из протестированных файлов не вернул результатов.")
def run(url, fetch_url, user_agent, detected_server):
if is_drupal_site(url, fetch_url, user_agent):
print(f"+ Обнаружен Drupal, продолжаем с перечислением модулей.")
module_path = "/sites/default/modules"
enumerate_drupal_modules(url, fetch_url, user_agent, module_path)
Ответ или решение
Для того чтобы полностью подавить сообщения об ошибках соединения, возникающие при использовании библиотеки requests
и urllib3
в Python, нам потребуется внести некоторые изменения в ваш код и конфигурацию. Давайте рассмотрим, как это можно сделать, шаг за шагом.
1. Отключение логирования соединений
Сначала вы правильно пробовали использовать модуль logging
, чтобы подавить сообщения от urllib3
и requests
. Убедитесь, что у вас правильно настроены уровни логирования:
import logging
# Установка уровня логирования для urllib3 и requests на CRITICAL
logging.getLogger("urllib3").setLevel(logging.CRITICAL)
logging.getLogger("requests").setLevel(logging.CRITICAL)
Эти строки кода установят наиболее высокий уровень логирования, и все сообщения уровня INFO, WARNING и ERROR не будут отображаться.
2. Отключение предупреждений
Кроме того, вы можете использовать модуль warnings
, чтобы подавить предупреждения:
import warnings
# Игнорирование предупреждений от urllib3
warnings.filterwarnings("ignore", category=UserWarning, module="urllib3")
Этот код должен игнорировать все предупреждения, но вы всё равно можете столкнуться с выводом ошибок в самом коде.
3. Обработка исключений в блоках try/except
Вы также можете попробовать обернуть ваши вызовы requests
в блоки try/except
, чтобы избежать отображения исключений на консоли. Например:
try:
response = fetch_url(full_url, method="GET", headers={"User-Agent": user_agent})
except requests.exceptions.RequestException:
# Обработка исключения, не выводя сообщение
return None # или любое другое значение, если нужно
4. Полное подавление Traceback
Чтобы полностью подавить вывод сообщения о ошибках и stack trace, вы можете использовать дополнительную обработку:
import sys
import traceback
def fetch_url_with_suppressed_errors(url, method="GET", **kwargs):
try:
return fetch_url(url, method=method, **kwargs)
except Exception as e:
# Печатаем ничего, вместо заполнения в консоли
return None # Либо любое другое значение, представляющее ошибку
# Используйте эту функцию вместо прямого вызова fetch_url
response = fetch_url_with_suppressed_errors(full_url, method="GET", headers={"User-Agent": user_agent})
5. Внедрение новой функции в ваш код
Убедитесь, что вы заменили все вызовы fetch_url
в вашем коде на новый метод fetch_url_with_suppressed_errors
. Это позволит вам избежать случайных ошибок и предотвратить вывод сообщений в консоль.
Заключение
Принимая во внимание вышеперечисленные шаги, вы сможете подавить вывод соединительных ошибок и исключений в Python, использующем библиотеки requests
и urllib3
. Если после выполнения этих действий ошибки продолжают отображаться, вероятно, необходимо проверить, где именно в коде происходит вывод. В полной мере придерживайтесь логирования и обработки исключений, и вы сможете контролировать поведение вашего приложения.
Не забудьте, что подавление всех ошибок может затруднить отладку, поэтому используйте этот подход с осторожностью, особенно в продуктивной среде.