core/security.py - master - HookDev-Arch/ServerMonitor

core/security.py
142 строки5 KB
HookDev-Arch
upload files
09 окт 2025, 02:25
09 окт 2025, 02:258bd6873
"""
Модуль безопасности для HookDev-Arch Dashboard v3.5
Включает шифрование паролей, валидацию данных и аутентификацию
"""
import hashlib
import secrets
import jwt
from datetime import datetime, timedelta
from pathlib import Path
from typing import Optional, Dict, Any
from cryptography.fernet import Fernet
import re


class PasswordManager:
    """Менеджер для безопасного хранения и обработки паролей"""
    
    def __init__(self, key_file: Path = None):
        self.key_file = key_file or Path("data/encryption.key")
        self.key = self._get_or_create_key()
        self.cipher = Fernet(self.key)
    
    def _get_or_create_key(self) -> bytes:
        """Получает или создает ключ шифрования"""
        if self.key_file.exists():
            return self.key_file.read_bytes()
        else:
            key = Fernet.generate_key()
            self.key_file.parent.mkdir(parents=True, exist_ok=True)
            self.key_file.write_bytes(key)
            return key
    
    def encrypt_password(self, password: str) -> str:
        """Шифрует пароль"""
        return self.cipher.encrypt(password.encode()).decode()
    
    def decrypt_password(self, encrypted: str) -> str:
        """Расшифровывает пароль"""
        return self.cipher.decrypt(encrypted.encode()).decode()


class DataValidator:
    """Валидатор входных данных для защиты от инъекций"""
    
    @staticmethod
    def validate_host(host: str) -> str:
        """Валидация IP адреса или домена"""
        # IP адрес
        ip_pattern = r'^(\d{1,3}\.){3}\d{1,3}$'
        if re.match(ip_pattern, host):
            parts = host.split('.')
            if all(0 <= int(part) <= 255 for part in parts):
                return host
        
        # Домен
        domain_pattern = r'^[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
        if re.match(domain_pattern, host):
            return host
        
        raise ValueError(f'Invalid host format: {host}')
    
    @staticmethod
    def validate_port(port: int) -> int:
        """Валидация порта"""
        if not 1 <= port <= 65535:
            raise ValueError(f'Port must be between 1 and 65535, got: {port}')
        return port
    
    @staticmethod
    def validate_name(name: str) -> str:
        """Валидация имени сервера"""
        if not name or len(name.strip()) < 2:
            raise ValueError('Server name must be at least 2 characters')
        
        # Удаляем потенциально опасные символы
        safe_name = re.sub(r'[<>:"/\\|?*]', '', name.strip())
        if not safe_name:
            raise ValueError('Server name contains only invalid characters')
        
        return safe_name
    
    @staticmethod
    def validate_user(user: str) -> str:
        """Валидация имени пользователя"""
        if not user or len(user.strip()) < 1:
            raise ValueError('Username cannot be empty')
        
        # Проверяем на безопасные символы
        if not re.match(r'^[a-zA-Z0-9._-]+$', user):
            raise ValueError('Username contains invalid characters')
        
        return user.strip()


class AuthenticationManager:
    """Менеджер аутентификации"""
    
    def __init__(self, secret_key: str = None):
        self.secret_key = secret_key or "hookdev-arch-secret-key-2024"
        self.algorithm = "HS256"
        self.token_expiry = timedelta(hours=24)
    
    def create_token(self, user_id: str, permissions: list = None) -> str:
        """Создает JWT токен"""
        payload = {
            'user_id': user_id,
            'permissions': permissions or ['read'],
            'exp': datetime.utcnow() + self.token_expiry,
            'iat': datetime.utcnow()
        }
        return jwt.encode(payload, self.secret_key, algorithm=self.algorithm)
    
    def verify_token(self, token: str) -> Dict[str, Any]:
        """Проверяет JWT токен"""
        try:
            payload = jwt.decode(token, self.secret_key, algorithms=[self.algorithm])
            return payload
        except jwt.ExpiredSignatureError:
            raise ValueError("Token has expired")
        except jwt.InvalidTokenError:
            raise ValueError("Invalid token")
    
    def hash_password(self, password: str) -> str:
        """Хэширует пароль для хранения"""
        salt = secrets.token_hex(16)
        hash_obj = hashlib.pbkdf2_hmac('sha256', password.encode(), salt.encode(), 100000)
        return f"{salt}:{hash_obj.hex()}"
    
    def verify_password(self, password: str, hashed: str) -> bool:
        """Проверяет пароль"""
        try:
            salt, hash_hex = hashed.split(':')
            hash_obj = hashlib.pbkdf2_hmac('sha256', password.encode(), salt.encode(), 100000)
            return hash_obj.hex() == hash_hex
        except ValueError:
            return False


# Глобальные экземпляры
password_manager = PasswordManager()
data_validator = DataValidator()
auth_manager = AuthenticationManager()