trevor/nas_control_bot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

init commit

Browse Source
This commit is contained in:
Choi A.K.
2025-08-30 10:33:46 +09:00
commit 49b3cea942
304 changed files with 116485 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

391
test_api_headers.py Normal file
View File

@@ -0,0 +1,391 @@
#!/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()