#!/usr/bin/env python # -*- coding: utf-8 -*- """ Тестовый скрипт для прямого доступа к API Synology для получения информации о системе. Используется для отладки и определения совместимых API. """ import requests import logging import json import sys import os import urllib3 from requests.adapters import HTTPAdapter from urllib3.util import Retry # Добавляем корневой каталог в путь для импорта sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from src.config.config import ( SYNOLOGY_HOST, SYNOLOGY_PORT, SYNOLOGY_USERNAME, SYNOLOGY_PASSWORD, SYNOLOGY_SECURE ) # Отключение предупреждений о небезопасных 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__) def direct_api_test(): """Прямой тест API без использования классов для определения проблемы""" # Создаем базовую сессию session = requests.Session() session.verify = False # Отключаем проверку SSL # Добавляем повторные попытки для HTTP-запросов 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: Авторизация logger.info("Шаг 1: Попытка авторизации...") # Сначала получаем информацию об API авторизации api_info_url = f"{base_url}/entry.cgi" api_info_params = { "api": "SYNO.API.Info", "version": "1", "method": "query", "query": "SYNO.API.Auth" } try: auth_info_response = session.get(api_info_url, params=api_info_params, timeout=10) 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}") # Пробуем версию 6 или максимальную доступную auth_version = min(6, auth_max_version) # Выполняем авторизацию 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": "DirectApiTest", "format": "cookie" } # Для версии 6+ используем немного другой формат 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, timeout=10) auth_data = auth_response.json() if auth_data.get("success"): sid = auth_data.get("data", {}).get("sid") logger.info(f"Авторизация успешна! SID: {sid[:10]}...") # Шаг 2: Тестирование различных API для получения информации о системе logger.info("Шаг 2: Тестирование различных API для получения информации о системе") # Создаем список API для тестирования api_to_test = [ {"name": "SYNO.DSM.Info", "method": "getinfo", "version": 1}, {"name": "SYNO.DSM.Info", "method": "getinfo", "version": 2}, {"name": "SYNO.Core.System", "method": "info", "version": 1}, {"name": "SYNO.Core.System", "method": "info", "version": 2}, {"name": "SYNO.Core.System.Status", "method": "get", "version": 1}, {"name": "SYNO.Core.System.Status", "method": "get", "version": 2}, {"name": "SYNO.Core.System.Utilization", "method": "get", "version": 1}, {"name": "SYNO.Core.CurrentConnection", "method": "list", "version": 1} ] # Перебираем все API и тестируем их for api in api_to_test: # Сначала получаем информацию о конкретном API try: api_info_params = { "api": "SYNO.API.Info", "version": "1", "method": "query", "query": api["name"] } api_info_resp = session.get(api_info_url, params=api_info_params, timeout=10) api_info_data = api_info_resp.json() if api_info_data.get("success") and api["name"] in api_info_data.get("data", {}): api_details = api_info_data["data"][api["name"]] api_path = api_details.get("path", "entry.cgi") api_min_version = api_details.get("minVersion", 1) api_max_version = api_details.get("maxVersion", 1) # Проверяем, поддерживается ли указанная версия if api["version"] < api_min_version: logger.warning(f"{api['name']} v{api['version']} ниже минимальной {api_min_version}, используем {api_min_version}") test_version = api_min_version elif api["version"] > api_max_version: logger.warning(f"{api['name']} v{api['version']} выше максимальной {api_max_version}, используем {api_max_version}") test_version = api_max_version else: test_version = api["version"] # Выполняем запрос API test_url = f"{base_url}/{api_path}" test_params = { "api": api["name"], "version": str(test_version), "method": api["method"], "_sid": sid # Используем sid для аутентификации } logger.info(f"Тестирование {api['name']}.{api['method']} v{test_version}") test_response = session.get(test_url, params=test_params, timeout=10) test_data = test_response.json() if test_data.get("success"): logger.info(f"API {api['name']}.{api['method']} v{test_version} РАБОТАЕТ!") logger.info(f"Результат: {json.dumps(test_data.get('data', {}), indent=2)[:200]}...") else: error_code = test_data.get("error", {}).get("code", -1) logger.error(f"API {api['name']}.{api['method']} v{test_version} ОШИБКА: {error_code}") # Если ошибка связана с сессией, попробуем еще раз авторизоваться if error_code == 119: # Session timeout logger.info("Повторная авторизация из-за ошибки 119...") # Создаем новую сессию new_session = requests.Session() new_session.verify = False auth_response = new_session.get(auth_url, params=auth_params, timeout=10) auth_data = auth_response.json() if auth_data.get("success"): new_sid = auth_data.get("data", {}).get("sid") logger.info(f"Повторная авторизация успешна! Новый SID: {new_sid[:10]}...") # Пробуем запрос с новым SID test_params["_sid"] = new_sid logger.info(f"Повторное тестирование {api['name']}.{api['method']} v{test_version}") test_response = new_session.get(test_url, params=test_params, timeout=10) test_data = test_response.json() if test_data.get("success"): logger.info(f"API {api['name']}.{api['method']} v{test_version} теперь РАБОТАЕТ!") logger.info(f"Результат с новой сессией: {json.dumps(test_data.get('data', {}), indent=2)[:200]}...") else: error_code = test_data.get("error", {}).get("code", -1) logger.error(f"API {api['name']}.{api['method']} v{test_version} ВСЕ ЕЩЕ С ОШИБКОЙ: {error_code}") else: logger.warning(f"API {api['name']} не найден в информации API") except Exception as e: logger.error(f"Ошибка при тестировании {api['name']}.{api['method']} v{api['version']}: {str(e)}") # Шаг 3: Тестирование комбинации запросов для решения проблемы logger.info("Шаг 3: Тестирование комбинации запросов для решения проблемы") # Создаем новую сессию для каждого запроса for api in [{"name": "SYNO.DSM.Info", "method": "getinfo", "version": 1}]: try: fresh_session = requests.Session() fresh_session.verify = False # Авторизуемся auth_response = fresh_session.get(auth_url, params=auth_params, timeout=10) auth_data = auth_response.json() if auth_data.get("success"): fresh_sid = auth_data.get("data", {}).get("sid") logger.info(f"Авторизация в новой сессии успешна! SID: {fresh_sid[:10]}...") # Сразу же делаем запрос для получения информации в той же сессии test_params = { "api": api["name"], "version": str(api["version"]), "method": api["method"], "_sid": fresh_sid } test_url = f"{base_url}/entry.cgi" # Используем entry.cgi по умолчанию logger.info(f"Тест в свежей сессии: {api['name']}.{api['method']} v{api['version']}") test_response = fresh_session.get(test_url, params=test_params, timeout=10) test_data = test_response.json() if test_data.get("success"): logger.info(f"API в свежей сессии РАБОТАЕТ!") logger.info(f"Результат: {json.dumps(test_data.get('data', {}), indent=2)[:200]}...") else: error_code = test_data.get("error", {}).get("code", -1) logger.error(f"API в свежей сессии ОШИБКА: {error_code}") except Exception as e: logger.error(f"Ошибка при тестировании свежей сессии: {str(e)}") # Шаг 4: Получаем информацию об остальных API logger.info("Шаг 4: Получаем информацию о доступных API для уточнения проблемы") # Запрашиваем все API из SYNO.API.Info try: all_api_params = { "api": "SYNO.API.Info", "version": "1", "method": "query", "query": "all" } all_api_response = session.get(api_info_url, params=all_api_params, timeout=15) # Больший таймаут для большого ответа all_api_data = all_api_response.json() if all_api_data.get("success"): api_list = all_api_data.get("data", {}) logger.info(f"Получен список всех API. Найдено {len(api_list)} API.") # Ищем интересующие нас API для отладки interested_in = ["SYNO.DSM.Info", "SYNO.Core.System", "SYNO.Core.Hardware", "SYNO.Core.System.Status", "SYNO.API.Auth"] logger.info("Информация о важных API:") for api_name in interested_in: if api_name in api_list: logger.info(f"{api_name}: {api_list[api_name]}") else: logger.warning(f"API {api_name} не найден") else: logger.error("Не удалось получить список всех API") except Exception as e: logger.error(f"Ошибка при получении списка API: {str(e)}") 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)}") if __name__ == "__main__": logger.info("Запуск прямого теста API Synology") direct_api_test()