Покрытие кода docstrings

This commit is contained in:
Merelden 2025-10-14 11:42:57 +05:00
parent 4e92aba2a2
commit 8f48df2cf5
19 changed files with 2324 additions and 1839 deletions

View file

@ -1,3 +1,27 @@
"""
Маршрутизация API-версии v1 для GONETS.
Модуль агрегирует и монтирует роутеры отдельных доменных областей под общим
корневым роутером. Каждый подключенный модуль предоставляет набор эндпоинтов
со своим префиксом и тегами для группировки в документации FastAPI.
Структура маршрутов:
- `/ufw` (UFW): управление правилами и состоянием Uncomplicated Firewall.
- `/ip` (IP): проверки и применение IP-конфигураций для сетевых профилей.
- `/network` (Network): получение и визуализация состояния интерфейсов,
управление позициями UI-контейнеров.
- `/auth` (Auth): регистрация, аутентификация (JWT), сессии.
- `/subnet` (Subnet): сканирование подсетей, заметки к устройствам,
взаимодействие с Wi‑Fi (скан/подключение/отключение/статус).
- `WebSocket`: обработчики взаимодействия по WS (без префикса, путь внутри
модуля), периодическая трансляция сетевых данных.
- `/settings` (Settings): сервисные операции над локальными JSON‑хранилищами
(очистка данных подсетей/wi‑fi и пр.).
Назначение:
- Централизованная точка подключения доменных роутеров к приложению.
- Единообразие тегов и префиксов для автогенерации схемы и документации.
"""
from fastapi import APIRouter
from .v1 import ufw, websocket, ip, network, auth, subnet, settings

View file

@ -15,8 +15,17 @@ class RuleNumbers(BaseModel):
@router.get("/rules", response_model=UFWRulesResponse)
async def get_ufw_rules():
"""
Получает все правила UFW с помощью сервиса UFWService и возвращает их в формате JSON.
"""Возвращает все правила UFW.
Осуществляет обращение к сервису `UFWService` для чтения текущего списка
правил брандмауэра и возвращает результат в сериализованном виде для клиента.
Returns:
UFWRulesResponse: Структура с коллекцией правил и сопутствующей
информацией о состоянии.
Raises:
HTTPException: В случае ошибок взаимодействия с UFW (status_code=500).
"""
ufw_service = UFWService()
try:
@ -27,8 +36,13 @@ async def get_ufw_rules():
@router.get("/status", response_model=UFWStatus)
async def get_ufw_status():
"""
Получает текущий статус UFW.
"""Возвращает текущий статус UFW.
Returns:
UFWStatus: Объект со сведениями о включенности UFW и деталями статуса.
Raises:
HTTPException: Если не удалось получить статус (status_code=500).
"""
ufw_service = UFWService()
try:
@ -39,8 +53,13 @@ async def get_ufw_status():
@router.post("/enable")
async def enable_ufw():
"""
Включает UFW.
"""Включает UFW.
Returns:
dict: Словарь с ключами `status` и `message` при успешном выполнении.
Raises:
HTTPException: Если включение завершилось неуспешно (status_code=500).
"""
ufw_service = UFWService()
try:
@ -56,8 +75,13 @@ async def enable_ufw():
@router.post("/disable")
async def disable_ufw():
"""
Выключает UFW.
"""Выключает UFW.
Returns:
dict: Словарь с ключами `status` и `message` при успешном выполнении.
Raises:
HTTPException: Если выключение завершилось неуспешно (status_code=500).
"""
ufw_service = UFWService()
try:
@ -73,8 +97,19 @@ async def disable_ufw():
@router.post("/rules/in")
async def add_ufw_in_rules(rules: List[UFWRule]):
"""
Добавляет входящие (in) правила UFW.
"""Добавляет входящие (in) правила UFW.
Args:
rules (List[UFWRule]): Список правил, определяющих направления,
порты/диапазоны/протоколы и действия (allow/deny/reject).
Returns:
dict: Итог выполнения с количеством успешно добавленных правил и деталями
по каждому правилу.
Raises:
HTTPException: Если UFW отключен (status_code=400) или при внутренних
ошибках (status_code=500).
"""
ufw_service = UFWService()
try:
@ -108,8 +143,17 @@ async def add_ufw_in_rules(rules: List[UFWRule]):
@router.post("/rules/out")
async def add_ufw_out_rules(rules: List[UFWRule]):
"""
Добавляет исходящие (out) правила UFW.
"""Добавляет исходящие (out) правила UFW.
Args:
rules (List[UFWRule]): Список правил исходящего трафика для добавления.
Returns:
dict: Объект статуса с сообщением об успешном добавлении.
Raises:
HTTPException: Если UFW отключен (status_code=400) или при внутренних
ошибках (status_code=500).
"""
ufw_service = UFWService()
try:
@ -137,8 +181,18 @@ async def add_ufw_out_rules(rules: List[UFWRule]):
@router.delete("/rules")
async def delete_ufw_rules(rule_numbers: RuleNumbers):
"""
Удаляет правила UFW по их номерам.
"""Удаляет правила UFW по их номерам.
Args:
rule_numbers (RuleNumbers): Номера правил, подлежащих удалению (как в
выводе `ufw status numbered`).
Returns:
dict: Информация об общем количестве удаляемых правил, числе успешных и
возможной ошибке.
Raises:
HTTPException: При ошибках удаления (status_code=500).
"""
ufw_service = UFWService()
try:

View file

@ -1,3 +1,20 @@
"""
Конфигурация CORS для FastAPI-приложения GONETS.
Модуль инкапсулирует добавление middleware `CORSMiddleware` со списком
разрешённых источников (origins), характерных для локальной разработки и
типовых портов фронтенда (например, Vite/React на 5173/5174, dev‑серверы).
Использование:
cors_config = CORSMiddlewareConfig(app)
cors_config.add_cors_middleware()
Поведение:
- Разрешены все методы (`allow_methods=["*"]`) и заголовки (`allow_headers=["*"]`).
- Разрешены креденшлы (`allow_credentials=True`) для работы с куками (локаль и др.).
- При необходимости список `origins` можно расширить или параметризовать
на основе конфигурационного файла окружения.
"""
from fastapi import FastAPI
from starlette.middleware.cors import CORSMiddleware

View file

@ -8,7 +8,18 @@ logger = logging.getLogger(__name__)
def init_config():
"""Создаёт конфигурационный файл с SECRET_KEY"""
"""
Инициализация конфигурации приложения (JWT SECRET_KEY).
Назначение:
- Создаёт файл `/etc/gonets/config.ini` с секцией `[JWT]` и случайным
`secret_key`, если он отсутствует. Устанавливает права `0o600`.
Замечания:
- Функция предназначена для однократного запуска (в инсталляторе/скрипте
подготовки окружения) и безопасно пропускает работу, если файл уже существует.
"""
"""Создаёт конфигурационный файл с SECRET_KEY."""
config_path = "/etc/gonets/config.ini"
config_dir = os.path.dirname(config_path)

View file

@ -2,7 +2,13 @@ from typing import Dict, List
class PortConfig:
"""Конфигурация портов для сетевого сканирования."""
"""Конфигурация известных портов для сетевого анализа.
Назначение:
Предоставляет сопоставление распространённых портов с человекочитаемыми
названиями сервисов для отображения в UI и логах. Поддерживает расширение
пользовательским словарём.
"""
_DEFAULT_PORTS: Dict[int, str] = {
20: "FTP (Data)",

View file

@ -3,8 +3,11 @@ from typing import Tuple
class CLIExecutor:
"""
Класс для выполнения CLI-команд
"""Утилита для безопасного выполнения CLI‑команд.
Назначение:
Инкапсулирует вызов `subprocess.run` и приводит результат к удобному
кортежу (код возврата, stdout) без генерации исключений при ошибках.
"""
@staticmethod

View file

@ -1,3 +1,28 @@
"""
Инициализация FastAPI-приложения GONETS.
Содержимое модуля:
- Создание экземпляра приложения FastAPI с метаданными (title, description, version).
- Подключение CORS-политики через `CORSMiddlewareConfig` с заранее заданными
доверенными источниками (localhost и стандартные порты фронтенда).
- Монтаж корневого роутера `api.router`, который инкапсулирует все `v1`-эндпоинты
(auth, subnet, ip, network, ufw, websocket, settings).
- Инициализация `NetworkService` на уровне приложения для подготовки сетевой
подсистемы (например, ленивые проверки окружения/зависимостей).
- Точка входа для локального запуска через `uvicorn` (режим разработки с reload).
Замечания по развертыванию:
- В прод-среде приложение запускается через `systemd` юнит `gonets.service`,
где в качестве команды используется `python3 SRV/main.py`.
- Для TLS можно использовать закомментированную строку с указанием
`ssl_keyfile` и `ssl_certfile` (сертификаты ожидаются в `/etc/gonets/certs`).
Безопасность и окружение:
- Перекрестные запросы (CORS) ограничены списком localhost-источников, что
соответствует типичной локальной разработке (Vite/DevServer и т.д.).
- Все пути и сервисы, требующие повышенных прав (например, системные команды),
должны выполняться в соответствующем окружении с необходимыми разрешениями.
"""
from fastapi import FastAPI
import uvicorn
from api.router import router

3738
SRV/poetry.lock generated

File diff suppressed because it is too large Load diff

View file

@ -1,3 +1,22 @@
"""
Сервис низкоуровневого ARP-сканирования локальной сети.
Назначение:
- Обнаружение активных узлов в пределах сетей хоста путём широковещательных
ARP‑запросов от имени выбранного интерфейса.
- Помощь сервисам более высокого уровня (например, IPService) при проверке
занятости адресов перед применением конфигураций.
Особенности реализации:
- Определяет подключённые сети интерфейса через `netifaces` и строит CIDR для
сканирования.
- Формирует ARP‑пакеты с помощью `scapy` (Ether/ARP) и отправляет их с таймаутом.
- Добавляет локальный адрес интерфейса в результирующий список клиентов.
Требования/ограничения:
- Для отправки ARP‑кадров могут потребоваться повышенные привилегии или
capability на сырой сокет.
"""
from fastapi import Depends
from scapy.layers.l2 import ARP, Ether
from scapy.sendrecv import srp

View file

@ -1,3 +1,19 @@
"""
Сервис управления подключениями Wi‑Fi через `nmcli`.
Назначение:
- Предоставляет операции подключения/отключения к точкам доступа и проверки
текущего статуса, ориентируясь на системный NetworkManager.
Поведение:
- Автоматически определяет активный Wi‑Fi‑интерфейс.
- Валидирует наличие SSID в листинге перед подключением и интерпретирует
частые ошибки `nmcli` (неверный пароль, отсутствующая сеть и т.п.).
Ограничения:
- Требуются права для управления сетевыми соединениями в системе.
- В окружениях без NetworkManager команды `nmcli` будут недоступны.
"""
import subprocess
import logging
from typing import Tuple, Optional

View file

@ -1,3 +1,16 @@
"""
Сервис для управления firewalld через D‑Bus.
Назначение:
- Обёртка над интерфейсами `org.fedoraproject.FirewallD1` для выполнения
типовых операций по зонам: просмотр/изменение дефолтной зоны, добавление и
удаление сервисов, портов, источников, перезагрузка конфигурации.
Примечания:
- Используется `pydbus.SystemBus` для доступа к системной шине.
- Методы генерируют дружелюбные исключения при ошибках, содержащие контекст
операции.
"""
from re import DEBUG
from typing import Dict, List, Optional

View file

@ -1,3 +1,17 @@
"""
Сервис проверки и применения IP‑настроек через NetworkManager (D‑Bus).
Назначение:
- Перед применением конфигурации проверяет занятость IP адреса, сканируя сегмент
сети средствами ARP (через `NetworkScanner`).
- Применяет изменения профиля подключения через `sdbus_async.networkmanager` и
выполняет переподключение интерфейса.
Требования/ограничения:
- Доступ к системной шине D‑Bus и права на изменения сетевых профилей.
- Структура входных данных должна соответствовать схеме `ipv4_updates` для
корректного обновления вложенных полей у профиля.
"""
from typing import Dict, Any
from sdbus_async.networkmanager import (
NetworkManager,

View file

@ -1,3 +1,16 @@
"""
Сервис чтения сетевого состояния через NetworkManager (D‑Bus).
Назначение:
- Предоставляет асинхронные методы для сборки унифицированных моделей
сетевых интерфейсов (Ethernet/Wi‑Fi), их IP‑конфигураций, а также получения
имени хоста. Используется API‑слоем для визуализации и управления.
Особенности:
- Фильтрация loopback и явно неактивных интерфейсов.
- Адаптация данных D‑Bus в pydantic‑модели (`NetworkInterface`, `WiFiDevice`,
`NetworkInterfacesContainer`).
"""
from typing import List, Dict, Tuple, Union, Optional
from sdbus_async.networkmanager import (
NetworkManager,

View file

@ -44,7 +44,20 @@ def get_subnet_service(translations: Translations = Depends(get_translations()))
class NetworkScanner:
"""Класс для сканирования подсети с использованием ARP-запросов.
Выполняет отправку ARP-запросов для обнаружения устройств в указанной подсети.
Назначение:
Обнаруживает активные устройства в локальной сети, отправляя широковещательные
ARP‑запросы по указанной подсети и собирая ответы (IP, MAC).
Детали реализации:
- Использует `scapy` (`Ether` + `ARP`) для генерации пакетов и `srp` для
отправки/получения на канальном уровне от имени заданного интерфейса.
- Таймаут ожидания ответа короткий (2 секунды), ориентирован на интерактивные
сценарии UI.
Ограничения и замечания:
- Требуются права для отправки низкоуровневых пакетов (capabilities/root).
- В сетях с фильтрацией широковещания количество обнаруженных устройств
может быть ограничено.
"""
def __init__(self, interface: str, translations: Translations):
"""Инициализирует сканер сети.
@ -74,6 +87,10 @@ class NetworkScanner:
Raises:
Exception: Если произошла ошибка сканирования.
"""
try:
arp = ARP(pdst=subnet)
except Exception as e:
logger.error(f"Error checking status: {str(e)}")
try:
arp = ARP(pdst=subnet)
ether = Ether(dst="ff:ff:ff:ff:ff:ff")
@ -87,7 +104,14 @@ class NetworkScanner:
class PortScanner:
"""Класс для сканирования открытых портов на устройствах.
Предоставляет функциональность для проверки доступности портов, пока не реализована.
Назначение:
Плейсхолдер для будущего функционала активного сканирования TCP/UDP‑портов
у обнаруженных узлов. В текущей версии возвращает пустой список.
План развития:
- Добавить неблокирующее сканирование наиболее распространённых портов
(22, 80, 443 и т.д.) с ограничением по времени.
- Настраиваемые профили глубины и скорости сканирования.
"""
def __init__(self, interface: str, translations: Translations):
"""Инициализирует сканер портов.
@ -117,7 +141,15 @@ class PortScanner:
class DeviceInfoCollector:
"""Класс для сбора информации об устройствах, включая вендоров и хостнеймы.
Использует внешние сервисы (MacLookup, NBTSCAN, DNS) для получения данных.
Источники данных:
- MacLookup: сопоставление префикса MAC (OUI) с названием производителя.
- DNS: `gethostbyaddr` с малым таймаутом для обратного разрешения имени.
- NetBIOS (nbtscan): опрос сегмента сети для извлечения NetBIOS‑имён.
Политика кэширования:
- Результаты `get_vendor` кэшируются с помощью `functools.lru_cache`.
- Результаты `nbtscan` аккумулируются в словарях `nbtscan_cache`/`nbtscan_results`
на время жизни экземпляра.
"""
def __init__(self, translations: Translations):
"""Инициализирует коллектор информации об устройствах.
@ -213,7 +245,15 @@ class DeviceInfoCollector:
class DeviceStorage:
"""Класс для хранения и управления данными об устройствах.
Обеспечивает загрузку и сохранение данных в JSON-файлы.
Формат файла подсети:
JSON‑структура вида
{
"devices": [Device, ...]
}
Политика очистки:
- Метод `_clean_old_devices` удаляет оффлайн‑устройства, не наблюдавшиеся
более 7 дней, чтобы поддерживать актуальность списка.
"""
def __init__(self, base_path: str = "storage/subnets", translations: Translations = None):
"""Инициализирует хранилище устройств.
@ -331,7 +371,19 @@ class DeviceStorage:
class SubnetService:
"""Сервис для сканирования подсетей и управления устройствами.
Объединяет функциональность сканирования, сбора информации и хранения данных.
Роль сервиса:
Объединяет сетевое сканирование (ARP), сбор дополнительной информации
(вендор, хостнейм), слияние с кэшем предыдущих результатов и персистентное
хранение данных по подсетям. Предоставляет вспомогательные операции для
локального узла и работы с заметками устройств.
Поток выполнения скана (`scan_subnet`):
1) Определение подсети (вручную или авто по интерфейсу).
2) Загрузка ранее сохранённых устройств подсети.
3) Добавление локального устройства (если принадлежит подсети).
4) ARP‑сканирование и параллельная обработка узлов (обогащение данных).
5) Маркировка оффлайн‑узлов из кэша, не встреченных в текущем скане.
6) Сортировка по IP и сохранение объединённого списка в JSON.
"""
def __init__(self, translations: Translations):
"""Инициализирует сервис подсети.

View file

@ -8,9 +8,16 @@ DEBUG = True
class UFWService:
"""
Сервис для взаимодействия с утилитой UFW (Uncomplicated Firewall) в Linux.
Позволяет получить список правил UFW и управлять ими.
"""Сервис взаимодействия с UFW (Uncomplicated Firewall).
Назначение:
- Чтение статуса и списка правил брандмауэра.
- Добавление, удаление и массовая обработка правил с поддержкой диапазонов портов,
комментариев и выбором протоколов.
Детали реализации:
- Команды выполняются через `sudo ufw ...`; вывод парсится в модели pydantic.
- Для операций удаления соблюдается порядок с конца, чтобы не нарушать нумерацию.
"""
def _run_command(self, command: str) -> str:

View file

@ -21,7 +21,17 @@ logging.basicConfig(
logger = logging.getLogger(__name__)
class WifiStorage:
"""Класс для хранения данных о Wi-Fi сетях и их заметках по SSID."""
"""Класс для хранения данных о Wi‑Fi сетях и их заметках по SSID.
Формат:
{
"networks": [{"ssid": str, "note": Optional[str]}]
}
Поведение:
- Создаёт директорию хранения при инициализации, ведёт логирование путей.
- Валидация структуры при чтении, безопасная запись с Unicode и отступами.
"""
def __init__(self, storage_path: str = "storage/wifi/wifi.json", translations: Translations = Depends(get_translations())):
self.storage_file = Path(storage_path)
self.storage_file.parent.mkdir(parents=True, exist_ok=True)
@ -61,7 +71,18 @@ class WifiStorage:
raise Exception(self.translations.gettext("Ошибка при сохранении заметок: {error}").format(error=str(e)))
class WifiService:
"""Сервис для сканирования текущих Wi-Fi точек доступа."""
"""Сервис для сканирования текущих Wi‑Fi точек доступа и заметок.
Источники данных:
- `pywifi` для базового сканирования и получения списка AP.
- `iw dev <iface> scan` для уточнения частоты, сигнала и безопасности.
- `MacLookup` для определения вендора по BSSID.
Политики:
- Выбор лучшего AP по SSID по уровню сигнала.
- Кэширование определения вендора.
- Сохранение SSID даже без заметок для будущих операций.
"""
def __init__(self, translations: Translations = Depends(get_translations())):
self.mac_lookup = MacLookup()
self.wifi = pywifi.PyWiFi()

View file

@ -11,8 +11,31 @@ logger = logging.getLogger(__name__)
class JsonStorage:
"""Хранилище пользователей в JSON-файле.
Предоставляет методы для создания, чтения и проверки данных пользователей,
используя шифрование паролей с помощью bcrypt.
Назначение:
Обеспечивает простое файловое хранилище для учетных записей пользователей
без полноценной СУБД. Подходит для встраиваемых окружений и сценариев,
где важна минимальная зависимость от внешних сервисов.
Формат данных:
JSON‑структура вида:
{
"users": [
{"id": int, "username": str, "password_hash": str, "is_admin": bool},
...
],
"next_id": int
}
Безопасность:
Пароли никогда не сохраняются в открытом виде. Для хеширования используется
`passlib.context.CryptContext` со схемой `bcrypt`. Файл создаётся с правами
доступа `0o600` (только владелец).
Ограничения и замечания:
- Отсутствует конкурентный доступ/блокировки файла — при использовании в
многопроцессной среде может потребоваться внешняя синхронизация.
- Имя файла и каталог по умолчанию `/etc/gonets/users.json` предполагают
наличие прав на запись и предварительную подготовку окружения.
"""
def __init__(self, file_path: str = "/etc/gonets/users.json"):

View file

@ -11,8 +11,33 @@ logger = logging.getLogger(__name__)
class SessionStorage:
"""Хранилище сессий пользователей в JSON-файле.
Предоставляет методы для создания, получения, удаления и очистки сессий с
учётом срока действия.
Назначение:
Лёгкое файловое хранилище для сессионных токенов, сопоставленных с
`client_id` (например, IP клиента). Предназначено для сред без БД и
persistent‑кэшей. Обеспечивает базовые операции и авто‑очистку протухших
сессий.
Формат данных:
{
"sessions": {
"<session_id>": {
"username": str,
"token": str,
"client_id": str,
"created_at": ISO8601 str,
"expiry": ISO8601 str
},
...
}
}
Политика TTL:
Каждая сессия действует 30 минут. При создании/чтении/удалении вызывается
`clean_expired_sessions()` для удаления протухших записей.
Ограничения:
- Нет файловых блокировок: в многопроцессных конфигурациях возможны гонки.
- Размещение по умолчанию `/etc/gonets/sessions.json` требует прав на запись.
"""
def __init__(self, file_path: str = "/etc/gonets/sessions.json"):

View file

@ -1,3 +1,21 @@
"""
Утилиты локализации и зависимости переводов для FastAPI.
Содержимое модуля:
- `get_translations()`: фабрика зависимости FastAPI, считывающая локаль из
query‑параметра `locale`, cookie `locale` или заголовка `Accept-Language`
(по приоритету), с дефолтом `ru`. Возвращает функцию‑зависимость, которая
подгружает `babel.support.Translations` из каталога `SRV/locales`.
- `_()`: удобная обёртка для получения перевода строки с использованием
текущей локали (или дефолтного `ru`, если контекст запроса не передан).
Особенности:
- При отсутствии словаря перевода для выбранной локали выполняется fallback на
русскую локаль, что гарантирует наличие строк интерфейса.
- Локаль определяется только по первому языку из `Accept-Language` (без
региональных суффиксов) для упрощения.
- Модуль ведёт отладочное логирование выбора и загрузки локали.
"""
from fastapi import Request, Depends
from babel.support import Translations
import logging