#!/usr/bin/env python # -*- coding: utf-8 -*- """ Тестовый скрипт для диагностики проблемы с использованием специальных заголовков """ import requests import logging import json import sys import os import urllib3 import time import socket from requests.adapters import HTTPAdapter from urllib3.util import Retry # Отключение предупреждений о небезопасных SSL-соединениях urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) # Настройка логирования logging.basicConfig( level=logging.DEBUG, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) logger = logging.getLogger(__name__) # Тестовые учетные данные (для примера) SYNOLOGY_HOST = "192.168.0.102" SYNOLOGY_PORT = 5000 SYNOLOGY_USERNAME = "superadmin" SYNOLOGY_PASSWORD = "Cl0ud_1985!" SYNOLOGY_SECURE = False SYNOLOGY_TIMEOUT = 10 def test_api_with_headers(): """Тестирование API с использованием специальных заголовков для решения проблемы 119""" # Создаем сессию session = requests.Session() session.verify = False # Отключаем проверку SSL # Настройки повторных попыток retry_strategy = Retry( total=3, status_forcelist=[429, 500, 502, 503, 504], allowed_methods=["GET", "POST"], backoff_factor=1.0 ) adapter = HTTPAdapter(max_retries=retry_strategy) session.mount("http://", adapter) session.mount("https://", adapter) # Формируем базовый URL protocol = "https" if SYNOLOGY_SECURE else "http" base_url = f"{protocol}://{SYNOLOGY_HOST}:{SYNOLOGY_PORT}/webapi" logger.info(f"Тестирование API с заголовками для {base_url}") # Тест 1: Получение SID с настройкой cookie и user-agent logger.info("Тест 1: Авторизация с настройкой cookie и user-agent") # Добавление пользовательских заголовков custom_headers = { 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36', 'Accept': 'application/json, text/javascript, */*; q=0.01', 'Accept-Language': 'ru-RU,ru;q=0.9,en-US;q=0.8,en;q=0.7', 'X-Requested-With': 'XMLHttpRequest', 'Connection': 'keep-alive', 'Referer': f'{protocol}://{SYNOLOGY_HOST}:{SYNOLOGY_PORT}/' } session.headers.update(custom_headers) try: # Определяем путь для авторизации auth_info_url = f"{base_url}/entry.cgi" auth_info_params = { "api": "SYNO.API.Info", "version": "1", "method": "query", "query": "SYNO.API.Auth" } auth_info_response = session.get(auth_info_url, params=auth_info_params) auth_info_data = auth_info_response.json() if auth_info_data.get("success"): auth_info = auth_info_data.get("data", {}).get("SYNO.API.Auth", {}) auth_path = auth_info.get("path", "auth.cgi") auth_max_version = auth_info.get("maxVersion", 6) logger.info(f"API авторизации: путь={auth_path}, макс. версия={auth_max_version}") # Используем версию 3 вместо 6 - тестирование на возможное решение проблемы auth_version = min(3, auth_max_version) # Пробуем более старую версию API # Выполняем авторизацию auth_url = f"{base_url}/{auth_path}" auth_params = { "api": "SYNO.API.Auth", "version": str(auth_version), "method": "login", "account": SYNOLOGY_USERNAME, "passwd": SYNOLOGY_PASSWORD, "session": "DirectHeaderTest", "format": "cookie" } if auth_version >= 6: auth_params["enable_syno_token"] = "yes" logger.info(f"Авторизация с использованием SYNO.API.Auth v{auth_version}") auth_response = session.get(auth_url, params=auth_params) auth_data = auth_response.json() if auth_data.get("success"): sid = auth_data.get("data", {}).get("sid") logger.info(f"Авторизация успешна! SID: {sid[:10]}...") # Теперь проверим, работает ли получение информации о системе # с настройкой cookie и заголовков # Сначала настроим куки для сохранения SID cookies = { 'id': sid, 'sid': sid } session.cookies.update(cookies) # Определяем путь для SYNO.DSM.Info info_info_url = f"{base_url}/entry.cgi" info_info_params = { "api": "SYNO.API.Info", "version": "1", "method": "query", "query": "SYNO.DSM.Info" } info_info_response = session.get(info_info_url, params=info_info_params) info_info_data = info_info_response.json() if info_info_data.get("success"): info_info = info_info_data.get("data", {}).get("SYNO.DSM.Info", {}) info_path = info_info.get("path", "entry.cgi") info_max_version = info_info.get("maxVersion", 1) info_min_version = info_info.get("minVersion", 1) logger.info(f"API SYNO.DSM.Info: путь={info_path}, версия={info_min_version}-{info_max_version}") # Используем правильную версию API info_version = min(2, info_max_version) # Делаем запрос для получения информации о системе # с использованием sid как параметр запроса logger.info("Тест 2: Получение информации о системе с SID как параметром") info_url = f"{base_url}/{info_path}" info_params = { "api": "SYNO.DSM.Info", "version": str(info_version), "method": "getinfo", "_sid": sid } logger.info(f"Запрос информации с использованием SYNO.DSM.Info v{info_version}") info_response = session.get(info_url, params=info_params) info_data = info_response.json() if info_data.get("success"): logger.info("Успешно получена информация о системе!") logger.info(f"Данные: {json.dumps(info_data.get('data', {}), indent=2)}") else: error_code = info_data.get("error", {}).get("code", -1) logger.error(f"Не удалось получить информацию о системе. Ошибка: {error_code}") # Пробуем альтернативный способ logger.info("Тест 3: Попытка получить базовую информацию через SYNO.Core.System") # Определяем путь для SYNO.Core.System system_info_url = f"{base_url}/entry.cgi" system_info_params = { "api": "SYNO.API.Info", "version": "1", "method": "query", "query": "SYNO.Core.System" } system_info_response = session.get(system_info_url, params=system_info_params) system_info_data = system_info_response.json() if system_info_data.get("success"): system_info = system_info_data.get("data", {}).get("SYNO.Core.System", {}) system_path = system_info.get("path", "entry.cgi") system_max_version = system_info.get("maxVersion", 1) system_min_version = system_info.get("minVersion", 1) logger.info(f"API SYNO.Core.System: путь={system_path}, версия={system_min_version}-{system_max_version}") # Используем правильную версию API system_version = 1 # Пробуем альтернативную стратегию с X-SYNO-TOKEN # Некоторые API Synology требуют специальный токен в заголовках token = auth_data.get("data", {}).get("synotoken") if token: session.headers.update({'X-SYNO-TOKEN': token}) logger.info(f"Добавлен X-SYNO-TOKEN: {token}") # Делаем запрос для получения информации о системе system_url = f"{base_url}/{system_path}" system_params = { "api": "SYNO.Core.System", "version": str(system_version), "method": "info", "_sid": sid } logger.info(f"Запрос информации с использованием SYNO.Core.System v{system_version}") system_response = session.get(system_url, params=system_params) system_data = system_response.json() if system_data.get("success"): logger.info("Успешно получена информация о системе через SYNO.Core.System!") logger.info(f"Данные: {json.dumps(system_data.get('data', {}), indent=2)}") else: error_code = system_data.get("error", {}).get("code", -1) logger.error(f"Не удалось получить информацию через SYNO.Core.System. Ошибка: {error_code}") # Пробуем другие методы для диагностики logger.info("Тест 4: Попытка использовать различные методы и заголовки") # Пробуем создать полностью новую сессию new_session = requests.Session() new_session.verify = False new_session.headers.update(custom_headers) # Пробуем версию 1 для авторизации auth_version = 1 auth_params["version"] = str(auth_version) auth_response = new_session.get(auth_url, params=auth_params) auth_data = auth_response.json() if auth_data.get("success"): sid = auth_data.get("data", {}).get("sid") logger.info(f"Новая авторизация (v{auth_version}) успешна! SID: {sid[:10]}...") # Добавляем SID как куки cookies = { 'id': sid, 'sid': sid } new_session.cookies.update(cookies) # Пробуем другой подход: разделение запросов во времени logger.info("Тест 5: Разделение запросов во времени") # Даем некоторое время для инициализации сессии на сервере time.sleep(2) # Пробуем получить список файлов filestation_info_url = f"{base_url}/entry.cgi" filestation_info_params = { "api": "SYNO.API.Info", "version": "1", "method": "query", "query": "SYNO.FileStation.List" } filestation_info_response = new_session.get(filestation_info_url, params=filestation_info_params) filestation_info_data = filestation_info_response.json() if filestation_info_data.get("success"): filestation_info = filestation_info_data.get("data", {}).get("SYNO.FileStation.List", {}) filestation_path = filestation_info.get("path", "entry.cgi") filestation_max_version = filestation_info.get("maxVersion", 1) logger.info(f"API SYNO.FileStation.List: путь={filestation_path}, макс. версия={filestation_max_version}") # Используем правильную версию API filestation_version = min(2, filestation_max_version) # Делаем запрос для получения списка общих папок filestation_url = f"{base_url}/{filestation_path}" filestation_params = { "api": "SYNO.FileStation.List", "version": str(filestation_version), "method": "list_share", "_sid": sid } logger.info(f"Запрос списка общих папок с использованием SYNO.FileStation.List v{filestation_version}") filestation_response = new_session.get(filestation_url, params=filestation_params) filestation_data = filestation_response.json() if filestation_data.get("success"): logger.info("Успешно получен список общих папок!") shares = filestation_data.get("data", {}).get("shares", []) logger.info(f"Общие папки: {json.dumps(shares, indent=2)[:200]}...") else: error_code = filestation_data.get("error", {}).get("code", -1) logger.error(f"Не удалось получить список общих папок. Ошибка: {error_code}") else: error_code = auth_data.get("error", {}).get("code", -1) logger.error(f"Новая авторизация не удалась! Код ошибки: {error_code}") else: logger.error("Не удалось получить информацию о SYNO.Core.System API") else: logger.error("Не удалось получить информацию о SYNO.DSM.Info API") else: error_code = auth_data.get("error", {}).get("code", -1) logger.error(f"Авторизация не удалась! Код ошибки: {error_code}") else: logger.error("Не удалось получить информацию об API авторизации") except Exception as e: logger.error(f"Произошла ошибка: {str(e)}") # Тест 6: Проверка сетевой доступности logger.info("Тест 6: Проверка сетевой доступности") try: # Проверка базового TCP-соединения socket_obj = socket.socket(socket.AF_INET, socket.SOCK_STREAM) socket_obj.settimeout(SYNOLOGY_TIMEOUT) result = socket_obj.connect_ex((SYNOLOGY_HOST, SYNOLOGY_PORT)) socket_obj.close() if result == 0: logger.info("TCP-соединение успешно установлено") else: logger.error(f"Не удалось установить TCP-соединение, код ошибки: {result}") except Exception as e: logger.error(f"Ошибка при проверке TCP-соединения: {str(e)}") # Тест 7: Запрос без аутентификации для проверки доступности API logger.info("Тест 7: Запрос без аутентификации для проверки доступности API") try: # Создаем новую сессию без аутентификации simple_session = requests.Session() simple_session.verify = False # Запрос к SYNO.API.Info не требует аутентификации api_info_url = f"{base_url}/entry.cgi" api_info_params = { "api": "SYNO.API.Info", "version": "1", "method": "query", "query": "all" } logger.info("Запрос информации о всех API без аутентификации") api_info_response = simple_session.get(api_info_url, params=api_info_params) if api_info_response.status_code == 200: logger.info("API доступно без аутентификации") api_info_data = api_info_response.json() if api_info_data.get("success"): logger.info("Успешно получена информация о всех API") api_count = len(api_info_data.get("data", {})) logger.info(f"Количество доступных API: {api_count}") # Поиск API для управления питанием power_apis = [] for api_name, api_info in api_info_data.get("data", {}).items(): if "power" in api_name.lower() or "reboot" in api_name.lower() or "shutdown" in api_name.lower(): power_apis.append(f"{api_name}: {api_info}") logger.info(f"Найдены API для управления питанием: {power_apis}") # Поиск API для получения информации о системе info_apis = [] for api_name, api_info in api_info_data.get("data", {}).items(): if "info" in api_name.lower() or "system" in api_name.lower() or "status" in api_name.lower(): info_apis.append(f"{api_name}: {api_info}") logger.info(f"Найдены API для информации о системе: {info_apis[:5]} и еще {len(info_apis)-5}") else: error_code = api_info_data.get("error", {}).get("code", -1) logger.error(f"Запрос к API без аутентификации не удался! Код ошибки: {error_code}") else: logger.error(f"API не доступно без аутентификации. HTTP статус: {api_info_response.status_code}") except Exception as e: logger.error(f"Ошибка при проверке доступности API: {str(e)}") if __name__ == "__main__": logger.info("Запуск теста API с заголовками") test_api_with_headers()