ДЕПОНИРУЕМЫЙ ЛИСТИНГ ИСХОДНОГО КОДА
CIFRA SECURE — Программный комплекс кибербезопасности и облачного мониторинга
============================================================================
Дата генерации: 2026-10-14 01:57 UTC
Модулей: 17
============================================================================
======================================================================МОДУЛЬ: Ядро лицензированияФАЙЛ: app/main.py======================================================================
# AV Platform Core — точка входа FastAPI
import os
from contextlib import asynccontextmanager
from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse, FileResponse, HTMLResponse
from fastapi.staticfiles import StaticFiles
from app.config import get_settings
from app.database import init_db
from app.routers import admin, cabinet, client, cloud, vault, recovery
from app.i18n import resolve_lang, t
settings = get_settings()
# Путь к статике (веб-панель)
STATIC_DIR = os.path.join(os.path.dirname(os.path.dirname(__file__)), "static")
@asynccontextmanager
async def lifespan(app: FastAPI):
"""Инициализация при старте: создание таблиц."""
await init_db()
yield
app = FastAPI(
title=settings.app_name,
description="Серверное ядро лицензирования и облачных сервисов (GLOBAL, без Geo-IP)",
version="1.0.0",
lifespan=lifespan,
)
# CORS (для веб-панели)
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# Глобальный обработчик ошибок для локализации
@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):
lang = resolve_lang(request.headers.get("accept-language"))
return JSONResponse(
status_code=500,
content={"detail": t("error.bad_request", lang)},
)
# Подключение роутеров
app.include_router(admin.router, prefix=settings.api_v1_prefix)
app.include_router(cabinet.router, prefix=settings.api_v1_prefix)
app.include_router(client.router, prefix=settings.api_v1_prefix)
app.include_router(cloud.router, prefix=settings.api_v1_prefix)
app.include_router(vault.router, prefix=settings.api_v1_prefix)
app.include_router(recovery.router, prefix=settings.api_v1_prefix)
@app.get("/health")
async def health():
"""Health-check эндпоинт."""
return {
"status": "ok",
"service": settings.app_name,
"region": "GLOBAL",
}
# ── Веб-панель (ESET HOME style) ──────────────────────────────
@app.get("/", include_in_schema=False)
async def root():
"""Корневой путь → личный кабинет (index.html)."""
index_path = os.path.join(STATIC_DIR, "index.html")
if os.path.exists(index_path):
return FileResponse(index_path, media_type="text/html")
# Фолбэк — JSON, если статика ещё не положена
return {
"service": settings.app_name,
"version": "1.0.0",
"region": "GLOBAL",
"region_label": "Global (Worldwide Activation)",
"docs": "/docs",
"api_prefix": settings.api_v1_prefix,
}
@app.get("/admin", include_in_schema=False)
async def admin_ui():
"""Админ-панель (admin.html)."""
admin_path = os.path.join(STATIC_DIR, "admin.html")
if os.path.exists(admin_path):
return FileResponse(admin_path, media_type="text/html")
return {"detail": "Admin UI not built yet."}
# ── Юридический пакет (Роспатент / ФИПС) ─────────────────────
LEGAL_DIR = os.path.join(os.path.dirname(STATIC_DIR), "legal")
LEGAL_DOCS = {
"eula": ("EULA_RU_EN.md", "Лицензионное соглашение (EULA) — RU/EN", "text/markdown; charset=utf-8"),
"privacy": ("PRIVACY_ZERO_TELEMETRY.md", "Политика конфиденциальности (Zero-Telemetry)", "text/markdown; charset=utf-8"),
"sbom": ("SBOM.md", "SBOM — перечень компонентов ПО", "text/markdown; charset=utf-8"),
"rospatent": ("ROSPATENT_PACKAGE.md", "Пакет данных для ФИПС (Роспатент)", "text/markdown; charset=utf-8"),
"deposition": ("source_code_deposition.txt", "Депонирование исходного кода", "text/plain; charset=utf-8"),
"listing": ("cifra_secure_listing.txt", "Листинг исходного кода (депозит)", "text/plain; charset=utf-8"),
}
@app.get("/legal", include_in_schema=False)
async def legal_index():
"""Индекс юридических документов CIFRA SECURE."""
rows = "".join(
f'{v[1]} — {v[0]}'
for k, v in LEGAL_DOCS.items()
)
html = f"""
CIFRA SECURE — Юридические документы
CIFRA SECURE — Юридические документы
Программный комплекс кибербезопасности и облачного мониторинга «CIFRA SECURE»
← На главную
"""
return HTMLResponse(html)
@app.get("/legal/{doc_key}", include_in_schema=False)
async def legal_doc(doc_key: str):
"""Отдать конкретный юридический документ."""
if doc_key not in LEGAL_DOCS:
return JSONResponse(status_code=404, content={"detail": "Документ не найден"})
filename, _, media = LEGAL_DOCS[doc_key]
path = os.path.join(LEGAL_DIR, filename)
if os.path.exists(path):
return FileResponse(path, media_type=media, filename=filename)
return JSONResponse(status_code=404, content={"detail": "Файл не смонтирован"})
# ── Скачивание агентов (привязано к кнопкам «Скачать защиту») ──
@app.get("/download/windows", include_in_schema=False)
async def download_windows():
"""Скачать Windows-агент.
Отдаёт собранный .exe (если есть) или архив с исходниками.
Правило: рекомендуется активировать ключ в день покупки.
"""
# Ищем собранный exe
exe_path = os.path.join(
os.path.dirname(STATIC_DIR), "windows-agent", "dist", "AVPlatformAgent.exe"
)
if os.path.exists(exe_path):
return FileResponse(
exe_path,
media_type="application/octet-stream",
filename="AVPlatformAgent.exe",
)
# Фолбэк — zip с исходниками агента
import shutil
import tempfile
agent_src = os.path.join(os.path.dirname(STATIC_DIR), "windows-agent")
if os.path.isdir(agent_src):
tmp = tempfile.mkdtemp()
zip_base = os.path.join(tmp, "AVPlatformAgent")
shutil.make_archive(zip_base, "zip", agent_src)
return FileResponse(
zip_base + ".zip",
media_type="application/zip",
filename="AVPlatformAgent.zip",
)
return JSONResponse(
status_code=404,
content={"detail": "Windows agent build not found."},
)
@app.get("/download/android", include_in_schema=False)
async def download_android():
"""Скачать Android-агента (плейсхолдер до сборки APK)."""
android_src = os.path.join(os.path.dirname(STATIC_DIR), "android-agent")
if os.path.isdir(android_src):
import shutil
import tempfile
tmp = tempfile.mkdtemp()
zip_base = os.path.join(tmp, "AVPlatformAgent-Android")
shutil.make_archive(zip_base, "zip", android_src)
return FileResponse(
zip_base + ".zip",
media_type="application/zip",
filename="AVPlatformAgent-Android.zip",
)
return JSONResponse(
status_code=404,
content={"detail": "Android agent not ready yet. Coming soon."},
)
# Раздача остальной статики (JS, манифест, иконки) из /static
if os.path.isdir(STATIC_DIR):
app.mount("/static", StaticFiles(directory=STATIC_DIR), name="static")
======================================================================МОДУЛЬ: Модели данныхФАЙЛ: app/models.py======================================================================
# SQLAlchemy ORM-модели (все таблицы)
import uuid
from datetime import datetime, timezone
from sqlalchemy import (
String, Integer, Boolean, DateTime, Text,
ForeignKey, Index, UniqueConstraint
)
from sqlalchemy.orm import Mapped, mapped_column, relationship
from app.database import Base
def _uuid() -> str:
return str(uuid.uuid4())
def _utcnow() -> datetime:
return datetime.now(timezone.utc)
class License(Base):
"""Лицензия (ключ) — основная сущность.
Все ключи ГЛОБАЛЬНЫЕ (region=GLOBAL, is_global=True).
Статус UNACTIVATED → ACTIVE (при первой активации) → EXPIRED.
Статус MERGED — ключ был объединён в другой (stack-key).
Статус REVOKED — ключ отозван админом.
"""
__tablename__ = "licenses"
id: Mapped[str] = mapped_column(String(36), primary_key=True, default=_uuid)
key_hash: Mapped[str] = mapped_column(String(64), unique=True, index=True)
# Полный ключ (нужен для Zero-Telemetry Recovery — возврат при восстановлении).
# В боевой версии рекомендуется шифровать at-rest серверным ключом.
key_plain: Mapped[str | None] = mapped_column(String(32), nullable=True)
key_masked: Mapped[str] = mapped_column(String(32))
edition: Mapped[str] = mapped_column(String(16)) # STANDARD / PLUS / PREMIUM
duration_days: Mapped[int] = mapped_column(Integer)
max_devices: Mapped[int] = mapped_column(Integer)
# Глобальный регион (без Geo-IP)
region: Mapped[str] = mapped_column(String(16), default="GLOBAL")
region_label: Mapped[str] = mapped_column(String(64), default="Global (Worldwide Activation)")
is_global: Mapped[bool] = mapped_column(Boolean, default=True)
status: Mapped[str] = mapped_column(String(16), default="UNACTIVATED", index=True)
activated_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
expires_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
# Настройки UI
ui_theme: Mapped[str] = mapped_column(String(8), default="dark") # dark / light
language: Mapped[str] = mapped_column(String(4), default="ru") # ru / en
# Восстановление ключа без Email (Zero-Telemetry Recovery)
# Хранит ТОЛЬКО SHA-256 (с солью av_recovery_v2_salt:) — сам код/фраза не хранятся.
recovery_hash: Mapped[str | None] = mapped_column(String(64), unique=True, index=True, nullable=True)
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_utcnow)
updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_utcnow, onupdate=_utcnow)
# Связи
devices: Mapped[list["Device"]] = relationship(
back_populates="license", cascade="all, delete-orphan"
)
invites: Mapped[list["Invite"]] = relationship(
back_populates="license", cascade="all, delete-orphan"
)
vault: Mapped["PasswordVault | None"] = relationship(
back_populates="license", cascade="all, delete-orphan", uselist=False
)
class Device(Base):
"""Устройство, привязанное к лицензии."""
__tablename__ = "devices"
__table_args__ = (
UniqueConstraint("license_id", "device_id", name="uq_license_device"),
)
id: Mapped[str] = mapped_column(String(36), primary_key=True, default=_uuid)
license_id: Mapped[str] = mapped_column(String(36), ForeignKey("licenses.id", ondelete="CASCADE"), index=True)
device_id: Mapped[str] = mapped_column(String(128)) # MachineGUID / Android ID
platform: Mapped[str] = mapped_column(String(16)) # WINDOWS / ANDROID
device_name: Mapped[str | None] = mapped_column(String(128), nullable=True)
is_active: Mapped[bool] = mapped_column(Boolean, default=True)
activated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_utcnow)
last_heartbeat_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
license: Mapped["License"] = relationship(back_populates="devices")
class Invite(Base):
"""Одноразовый инвайт-код для привязки устройств без раскрытия мастер-ключа."""
__tablename__ = "invites"
id: Mapped[str] = mapped_column(String(36), primary_key=True, default=_uuid)
license_id: Mapped[str] = mapped_column(String(36), ForeignKey("licenses.id", ondelete="CASCADE"), index=True)
token_hash: Mapped[str] = mapped_column(String(64), unique=True, index=True)
token_masked: Mapped[str] = mapped_column(String(32))
max_uses: Mapped[int] = mapped_column(Integer, default=1)
uses_count: Mapped[int] = mapped_column(Integer, default=0)
is_active: Mapped[bool] = mapped_column(Boolean, default=True)
region: Mapped[str] = mapped_column(String(16), default="GLOBAL")
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_utcnow)
expires_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
license: Mapped["License"] = relationship(back_populates="invites")
class PasswordVault(Base):
"""Хранилище паролей (Zero-Knowledge, зашифрованный слепок)."""
__tablename__ = "password_vaults"
id: Mapped[str] = mapped_column(String(36), primary_key=True, default=_uuid)
license_id: Mapped[str] = mapped_column(String(36), ForeignKey("licenses.id", ondelete="CASCADE"), unique=True, index=True)
encrypted_blob: Mapped[str] = mapped_column(Text)
blob_hash: Mapped[str] = mapped_column(String(64)) # SHA-256 для целостности
entry_count: Mapped[int] = mapped_column(Integer, default=0)
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_utcnow)
updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_utcnow, onupdate=_utcnow)
license: Mapped["License"] = relationship(back_populates="vault")
class ThreatSignatures(Base):
"""Таблица сигнатур угроз для облачного лукапа."""
__tablename__ = "threat_signatures"
id: Mapped[str] = mapped_column(String(36), primary_key=True, default=_uuid)
sha256: Mapped[str] = mapped_column(String(64), unique=True, index=True)
threat_name: Mapped[str] = mapped_column(String(256))
severity: Mapped[str] = mapped_column(String(16), default="HIGH") # LOW / MEDIUM / HIGH / CRITICAL
family: Mapped[str | None] = mapped_column(String(128), nullable=True)
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_utcnow)
class SpamPhones(Base):
"""База спам-номеров (Who Calls / Антиспам звонков).
category: SCAM / BANK_SPAM / ADS / SAFE
Хранится нормализованный номер или префикс (для RU +7..., международные).
"""
__tablename__ = "spam_phones"
id: Mapped[str] = mapped_column(String(36), primary_key=True, default=_uuid)
phone_normalized: Mapped[str] = mapped_column(String(32), unique=True, index=True)
category: Mapped[str] = mapped_column(String(16), default="SCAM") # SCAM / BANK_SPAM / ADS / SAFE
description_ru: Mapped[str | None] = mapped_column(String(256), nullable=True)
description_en: Mapped[str | None] = mapped_column(String(256), nullable=True)
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_utcnow)
class KeyMergeHistory(Base):
"""История наложения/объединения ключей (stack-key)."""
__tablename__ = "key_merge_history"
id: Mapped[str] = mapped_column(String(36), primary_key=True, default=_uuid)
target_license_id: Mapped[str] = mapped_column(String(36), ForeignKey("licenses.id", ondelete="CASCADE"), index=True)
source_license_id: Mapped[str] = mapped_column(String(36), ForeignKey("licenses.id", ondelete="CASCADE"), index=True)
source_key_masked: Mapped[str] = mapped_column(String(32))
old_expires_at: Mapped[datetime] = mapped_column(DateTime(timezone=True))
new_expires_at: Mapped[datetime] = mapped_column(DateTime(timezone=True))
added_days: Mapped[int] = mapped_column(Integer)
old_edition: Mapped[str] = mapped_column(String(16))
new_edition: Mapped[str] = mapped_column(String(16))
old_max_devices: Mapped[int] = mapped_column(Integer)
new_max_devices: Mapped[int] = mapped_column(Integer)
merged_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_utcnow)
======================================================================МОДУЛЬ: Схемы APIФАЙЛ: app/schemas.py======================================================================
# Pydantic v2 — схемы запросов и ответов
from datetime import datetime
from typing import Literal
from pydantic import BaseModel, Field, ConfigDict
# ===================== ENUM-константы =====================
# Новая коммерческая сетка 5 тарифов + алиасы старых 3 (для совместимости тестов)
EditionLiteral = Literal[
"ANTIVIRUS", # AVIR- (алиас STANDARD)
"INTERNET_SECURITY", # ISEC- (алиас PLUS)
"TOTAL_SECURITY", # TOTL- (алиас PREMIUM)
"ULTIMATE", # ULTM-
"SMALL_OFFICE", # BIZS-
"STANDARD", "PLUS", "PREMIUM", # алиасы
]
PlatformLiteral = Literal["WINDOWS", "ANDROID"]
ThemeLiteral = Literal["dark", "light"]
LanguageLiteral = Literal["ru", "en"]
StatusLiteral = Literal["UNACTIVATED", "ACTIVE", "EXPIRED", "MERGED", "REVOKED"]
HashStatusLiteral = Literal["CLEAN", "MALWARE", "UNKNOWN"]
# ===================== Базовые ответы =====================
class RegionMixin(BaseModel):
"""Mixin: все ответы с ключом содержат ГЛОБАЛЬНЫЙ регион."""
region: Literal["GLOBAL"] = "GLOBAL"
region_label: str = "Global (Worldwide Activation)"
is_global: bool = True
# ===================== Лицензия =====================
class LicenseOut(RegionMixin):
"""Ответ с данными лицензии."""
model_config = ConfigDict(from_attributes=True)
id: str
key_masked: str
edition: EditionLiteral
duration_days: int
max_devices: int
status: StatusLiteral
activated_at: datetime | None
expires_at: datetime | None
ui_theme: ThemeLiteral
language: LanguageLiteral
created_at: datetime
class LicenseCreatedOut(RegionMixin):
"""Ответ при генерации нового ключа (админ)."""
id: str
license_key: str # Полный ключ — возвращается ТОЛЬКО при генерации
key_masked: str
edition: EditionLiteral
duration_days: int
max_devices: int
status: StatusLiteral
# ===================== Админ: генерация ключей =====================
class GenerateKeyRequest(BaseModel):
"""Запрос на генерацию нового ключа."""
edition: EditionLiteral = "STANDARD"
duration_days: int = Field(default=365, ge=1, le=3650)
max_devices: int = Field(default=1, ge=1, le=999)
count: int = Field(default=1, ge=1, le=100)
class GenerateKeyResponse(BaseModel):
"""Ответ со списком сгенерированных ключей."""
keys: list[LicenseCreatedOut]
generated: int
# ===================== Активация =====================
class ActivateRequest(BaseModel):
"""Запрос активации по ключу или инвайту."""
license_key: str | None = None
invite_token: str | None = None
device_id: str = Field(..., min_length=1, max_length=128)
platform: PlatformLiteral
device_name: str | None = Field(default=None, max_length=128)
class ActivateResponse(RegionMixin):
"""Ответ активации."""
license: LicenseOut
device_id: str
message: str
# ===================== Heartbeat =====================
class HeartbeatRequest(BaseModel):
"""Запрос heartbeat (проверка подписки)."""
license_key: str
device_id: str
class HeartbeatResponse(RegionMixin):
"""Ответ heartbeat."""
status: StatusLiteral
expires_at: datetime | None
days_left: int | None
message: str
# ===================== Устройства =====================
class DeviceOut(BaseModel):
"""Данные устройства."""
model_config = ConfigDict(from_attributes=True)
id: str
device_id: str
platform: PlatformLiteral
device_name: str | None
is_active: bool
activated_at: datetime
last_heartbeat_at: datetime | None
class DeviceListOut(BaseModel):
"""Список устройств."""
devices: list[DeviceOut]
total: int
max_devices: int
class UnbindRequest(BaseModel):
"""Запрос отвязки устройства."""
license_key: str
device_id: str
class UnbindResponse(BaseModel):
"""Ответ отвязки."""
success: bool
message: str
# ===================== Cabinet: stack-key =====================
class StackKeyRequest(BaseModel):
"""Запрос на наложение/объединение ключей."""
current_key: str
new_key: str
class StackKeyResponse(RegionMixin):
"""Ответ stack-key."""
license: LicenseOut
added_days: int
old_expires_at: datetime
new_expires_at: datetime
edition_upgraded: bool
devices_upgraded: bool
message: str
# ===================== Cabinet: switch-key (upgrade/downgrade) =====================
class SwitchKeyRequest(BaseModel):
"""Запрос двустороннего переключения тарифа (Upgrade & Downgrade)."""
current_key: str
new_key: str
class SwitchKeyResponse(RegionMixin):
"""Ответ switch-key."""
license: LicenseOut
direction: Literal["upgrade", "downgrade"]
old_edition: str
new_edition: str
added_days: int
new_expires_at: datetime
vault_read_only: bool
message: str
# ===================== Cabinet: preferences =====================
class PreferencesRequest(BaseModel):
"""Запрос сохранения настроек UI."""
ui_theme: ThemeLiteral | None = None
language: LanguageLiteral | None = None
class PreferencesResponse(RegionMixin):
"""Ответ с обновлёнными настройками."""
ui_theme: ThemeLiteral
language: LanguageLiteral
message: str
# ===================== Cloud: hash lookup =====================
class CloudLookupRequest(BaseModel):
"""Запрос облачной проверки хэшей."""
hashes: list[str] = Field(..., min_length=1, max_length=100)
class HashResult(BaseModel):
"""Результат проверки одного хэша."""
sha256: str
status: HashStatusLiteral
threat_name: str | None
class CloudLookupResponse(RegionMixin):
"""Ответ облачного лукапа."""
results: list[HashResult]
checked: int
# ===================== Vault: синхронизация =====================
class VaultSyncRequest(BaseModel):
"""Запрос загрузки зашифрованного слепка паролей."""
encrypted_blob: str
blob_hash: str = Field(..., min_length=64, max_length=64)
entry_count: int = Field(..., ge=0)
class VaultSyncOut(RegionMixin):
"""Ответ синхронизации хранилища."""
encrypted_blob: str | None
blob_hash: str | None
entry_count: int
updated_at: datetime | None
message: str
# ===================== Приглашения (инвайты) =====================
class CreateInviteRequest(BaseModel):
"""Запрос создания инвайт-кода."""
license_key: str
max_uses: int = Field(default=1, ge=1, le=999)
class InviteOut(RegionMixin):
"""Ответ с данными инвайта."""
id: str
token_masked: str
max_uses: int
uses_count: int
is_active: bool
created_at: datetime
expires_at: datetime | None
class CreateInviteResponse(RegionMixin):
"""Ответ при создании инвайта (токен возвращается ТОЛЬКО здесь)."""
invite: InviteOut
invite_token: str # Полный токен — возвращается только при создании
message: str
# ===================== Health =====================
class HealthOut(BaseModel):
"""Ответ health-check."""
status: str
service: str
region: str
# ===================== Спам-номера (Who Calls / Антиспам) =====================
class PhoneAddRequest(BaseModel):
"""Запрос добавления спам-номера в базу."""
phone: str = Field(..., min_length=3, max_length=32)
category: Literal["SCAM", "BANK_SPAM", "ADS", "SAFE"] = "SCAM"
description_ru: str | None = Field(default=None, max_length=256)
description_en: str | None = Field(default=None, max_length=256)
class PhoneLookupRequest(BaseModel):
"""Запрос проверки входящего номера (Who Calls)."""
phone: str = Field(..., min_length=3, max_length=32)
class PhoneLookupResponse(RegionMixin):
"""Ответ проверки номера."""
phone: str
status: Literal["CLEAN", "SPAM"]
category: str | None
description: str | None
message: str
# ===================== Восстановление ключа (Zero-Telemetry Recovery) =====================
class RestoreByHwidRequest(BaseModel):
"""Запрос авто-восстановления по железу (HWID)."""
device_id: str = Field(..., min_length=1, max_length=128)
platform: PlatformLiteral = "WINDOWS"
class GenerateRecoveryCodeResponse(BaseModel):
"""Ответ: сгенерированное число восстановления (показывается 1 раз)."""
code: str
message: str
class SetRecoveryHashRequest(BaseModel):
"""Запрос: установить собственную связку (слово + 4 цифры)."""
secret_word: str = Field(..., min_length=3, max_length=64)
pin_digits: str = Field(..., min_length=4, max_length=4, pattern=r"^\d{4}$")
class RecoverKeyRequest(BaseModel):
"""Запрос восстановления: ИЛИ code (XXX-XXX-XX), ИЛИ secret_word + pin_digits."""
code: str | None = Field(default=None, max_length=32)
secret_word: str | None = Field(default=None, max_length=64)
pin_digits: str | None = Field(default=None, max_length=4)
class RecoverKeyResponse(RegionMixin):
"""Ответ восстановления: полный ключ + профиль лицензии."""
license_key: str
edition: str
duration_days: int
max_devices: int
status: str
expires_at: datetime | None
message: str
======================================================================МОДУЛЬ: Восстановление ключаФАЙЛ: app/routers/recovery.py======================================================================
# -*- coding: utf-8 -*-
"""Роутер: восстановление ключа без Email (Zero-Telemetry Recovery).
Эндпоинты:
POST /client/restore-by-hwid — авто-восстановление по железу (0 кликов)
POST /cabinet/generate-recovery-code — сгенерировать число XXX-XXX-XX (Вариант А)
POST /cabinet/set-recovery-hash — своя связка слово+4 цифры (Вариант Б)
POST /cabinet/recover-key — восстановить по коду/связке (rate-limit Redis)
GET /cabinet/recovery-kit.txt — скачать памятку с ключом
"""
from fastapi import APIRouter, Depends, HTTPException, Request
from fastapi.responses import PlainTextResponse
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from redis.asyncio import Redis
from app.database import get_db, get_redis
from app.deps import get_language, get_license_by_key_header
from app.i18n import LanguageCode, t
from app.models import License, Device
from app.schemas import (
RestoreByHwidRequest,
GenerateRecoveryCodeResponse,
SetRecoveryHashRequest,
RecoverKeyRequest,
RecoverKeyResponse,
)
from app.services import recovery_service
router = APIRouter(tags=["recovery"])
def _now_active(lic: License) -> bool:
"""Лицензия активна и не истекла."""
from datetime import datetime, timezone
if lic.status != "ACTIVE":
return False
if lic.expires_at is None:
return True
return lic.expires_at > datetime.now(timezone.utc)
@router.post("/client/restore-by-hwid", response_model=RecoverKeyResponse)
async def restore_by_hwid(
payload: RestoreByHwidRequest,
db: AsyncSession = Depends(get_db),
lang: LanguageCode = Depends(get_language),
):
"""Авто-восстановление по железу: hwid → активная лицензия → вернуть ключ."""
# hwid может встречаться у нескольких лицензий — берём последнее активное устройство
result = await db.execute(
select(Device)
.where(Device.device_id == payload.device_id,
Device.platform == payload.platform)
.order_by(Device.activated_at.desc())
)
devices = result.scalars().all()
if not devices:
raise HTTPException(status_code=404, detail=t("recovery.not_found", lang))
lic = None
for dev in devices:
lic_result = await db.execute(select(License).where(License.id == dev.license_id))
candidate = lic_result.scalar_one_or_none()
if candidate is not None and _now_active(candidate):
lic = candidate
break
if lic is None:
raise HTTPException(status_code=404, detail=t("recovery.not_found", lang))
return _recovery_response(lic, t("recovery.restored_by_hwid", lang))
@router.post("/cabinet/generate-recovery-code", response_model=GenerateRecoveryCodeResponse)
async def generate_recovery_code(
db: AsyncSession = Depends(get_db),
redis_client: Redis = Depends(get_redis),
license_obj: License = Depends(get_license_by_key_header),
lang: LanguageCode = Depends(get_language),
):
"""Вариант А: сгенерировать уникальное число XXX-XXX-XX (показать 1 раз)."""
for _ in range(20): # защита от коллизий
code = recovery_service.generate_numeric_code()
h = recovery_service.hash_recovery(code)
# Проверка коллизии: хэш не занят другим ключом
existing = await db.execute(
select(License).where(License.recovery_hash == h,
License.id != license_obj.id)
)
if existing.scalar_one_or_none() is None:
license_obj.recovery_hash = h
await db.commit()
return GenerateRecoveryCodeResponse(
code=code,
message=t("recovery.code_generated", lang),
)
raise HTTPException(status_code=500, detail=t("recovery.code_fail", lang))
@router.post("/cabinet/set-recovery-hash")
async def set_recovery_hash(
payload: SetRecoveryHashRequest,
db: AsyncSession = Depends(get_db),
license_obj: License = Depends(get_license_by_key_header),
lang: LanguageCode = Depends(get_language),
):
"""Вариант Б: своя связка (слово + 4 цифры). 409 если хэш занят."""
h = recovery_service.build_phrase_hash(payload.secret_word, payload.pin_digits)
existing = await db.execute(
select(License).where(License.recovery_hash == h,
License.id != license_obj.id)
)
if existing.scalar_one_or_none() is not None:
raise HTTPException(status_code=409, detail=t("recovery.hash_taken", lang))
license_obj.recovery_hash = h
await db.commit()
return {"message": t("recovery.hash_set", lang)}
@router.post("/cabinet/recover-key", response_model=RecoverKeyResponse)
async def recover_key(
payload: RecoverKeyRequest,
request: Request,
db: AsyncSession = Depends(get_db),
redis_client: Redis = Depends(get_redis),
lang: LanguageCode = Depends(get_language),
):
"""Восстановить ключ по коду XXX-XXX-XX ИЛИ связке слово+цифры. Rate-limit 5/10мин."""
# Определяем клиентский идентификатор для rate-limit (по IP)
client_id = request.client.host if request.client else "unknown"
if not await recovery_service.check_rate_limit(redis_client, client_id):
raise HTTPException(status_code=429, detail=t("recovery.rate_limited", lang))
# Формируем хэш из ввода
if payload.code:
h = recovery_service.hash_recovery(recovery_service.normalize_code(payload.code))
elif payload.secret_word and payload.pin_digits:
h = recovery_service.build_phrase_hash(payload.secret_word, payload.pin_digits)
else:
raise HTTPException(status_code=400, detail=t("recovery.bad_input", lang))
await recovery_service.register_attempt(redis_client, client_id)
result = await db.execute(select(License).where(License.recovery_hash == h))
lic = result.scalar_one_or_none()
if lic is None or not _now_active(lic):
raise HTTPException(status_code=404, detail=t("recovery.not_found", lang))
remaining = await recovery_service.remaining_attempts(redis_client, client_id)
resp = _recovery_response(lic, t("recovery.restored", lang))
return resp
@router.get("/cabinet/recovery-kit.txt")
async def recovery_kit(
db: AsyncSession = Depends(get_db),
license_obj: License = Depends(get_license_by_key_header),
):
"""Скачать памятку с ключом (.txt)."""
lines = [
"=== CIFRA SECURE | СЕРТИФИКАТ ЛИЦЕНЗИИ И ВОССТАНОВЛЕНИЯ ===",
"",
"Ваш ключ активации (сохраните в надёжном месте):",
f" {license_obj.key_plain or license_obj.key_masked}",
"",
"Способы восстановления ключа (без Email, 100% анонимно):",
" 1) Авто-по железу (HWID): запустите агент без ключа на том же ПК.",
" 2) По числу восстановления XXX-XXX-XX (если генерировали).",
" 3) По связке: ваше слово + 4 цифры (если задавали).",
"",
"⚠️ ПАМЯТКА: Обязательно сохраните свой ключ активации и (или)",
"число/фразу восстановления! Из-за политики 100% анонимности",
"(Zero-Telemetry) мы не храним ваши личные данные.",
"",
"Активируйте ключ в день покупки.",
"CIFRA SECURE • Global Edition",
]
return PlainTextResponse(
"\n".join(lines),
media_type="text/plain; charset=utf-8",
headers={"Content-Disposition": "attachment; filename=CIFRA_SECURE_KEY_BACKUP.txt"},
)
def _recovery_response(lic: License, message: str) -> RecoverKeyResponse:
return RecoverKeyResponse(
license_key=lic.key_plain or lic.key_masked,
edition=lic.edition,
duration_days=lic.duration_days,
max_devices=lic.max_devices,
status=lic.status,
expires_at=lic.expires_at,
message=message,
region="GLOBAL",
region_label="Global (Worldwide Activation)",
is_global=True,
)
======================================================================МОДУЛЬ: Лицензионный сервисФАЙЛ: app/services/license_service.py======================================================================
# Сервис лицензий: активация, heartbeat, stack-key, switch-key, инвайты
from datetime import datetime, timedelta, timezone
from sqlalchemy import select, func
from sqlalchemy.ext.asyncio import AsyncSession
from app.models import (
License, Device, Invite, KeyMergeHistory,
)
from app.security import (
generate_license_key, generate_invite_token,
hash_key, mask_key, mask_token,
)
from app.config import get_settings
settings = get_settings()
# Ранг тарифов (для upgrade/downgrade и stack-key).
# Алиасы старых тарифов маппим на новые.
EDITION_RANK = {
"ANTIVIRUS": 1, "STANDARD": 1,
"INTERNET_SECURITY": 2, "PLUS": 2,
"TOTAL_SECURITY": 3, "PREMIUM": 3,
"ULTIMATE": 4,
"SMALL_OFFICE": 5,
}
def canonical_edition(edition: str) -> str:
"""Алиас → каноническое имя нового тарифа (для отображения)."""
return {
"STANDARD": "ANTIVIRUS",
"PLUS": "INTERNET_SECURITY",
"PREMIUM": "TOTAL_SECURITY",
}.get(edition, edition)
def _utcnow() -> datetime:
return datetime.now(timezone.utc)
async def create_license(
db: AsyncSession,
edition: str,
duration_days: int,
max_devices: int,
) -> tuple[License, str]:
"""Создать новый ключ (UNACTIVATED, GLOBAL)."""
raw_key = generate_license_key(edition)
key_hash = hash_key(raw_key)
license_obj = License(
key_hash=key_hash,
key_plain=raw_key,
key_masked=mask_key(raw_key),
edition=edition,
duration_days=duration_days,
max_devices=max_devices,
region="GLOBAL",
region_label="Global (Worldwide Activation)",
is_global=True,
status="UNACTIVATED",
)
db.add(license_obj)
await db.commit()
await db.refresh(license_obj)
return license_obj, raw_key
async def get_license_by_raw_key(
db: AsyncSession,
raw_key: str,
) -> License | None:
"""Найти лицензию по сырому ключу."""
key_hash = hash_key(raw_key)
result = await db.execute(
select(License).where(License.key_hash == key_hash)
)
return result.scalar_one_or_none()
async def activate_license(
db: AsyncSession,
license_key: str | None,
invite_token: str | None,
device_id: str,
platform: str,
device_name: str | None,
) -> tuple[License, Device]:
"""Активировать лицензию по ключу или инвайту.
- Если UNACTIVATED → ставит activated_at=now, expires_at=now+duration_days, статус=ACTIVE.
- Если ACTIVE → проверяет лимит слотов и добавляет устройство.
- Если EXPIRED / REVOKED / MERGED → ошибка.
- Если устройство уже привязано → обновляет heartbeat.
"""
if license_key:
license_obj = await get_license_by_raw_key(db, license_key)
if license_obj is None:
raise ValueError("license.not_found")
invite_obj = None
elif invite_token:
invite_hash = hash_key(invite_token)
invite_result = await db.execute(
select(Invite).where(Invite.token_hash == invite_hash)
)
invite_obj = invite_result.scalar_one_or_none()
if invite_obj is None:
raise ValueError("invite.not_found")
if not invite_obj.is_active:
raise ValueError("invite.used_up")
if invite_obj.expires_at and invite_obj.expires_at < _utcnow():
raise ValueError("invite.expired")
license_obj = await db.get(License, invite_obj.license_id)
if license_obj is None:
raise ValueError("license.not_found")
else:
raise ValueError("auth.key_required")
# Проверка статуса лицензии
if license_obj.status == "REVOKED":
raise ValueError("license.revoked")
if license_obj.status == "MERGED":
raise ValueError("license.merged")
if license_obj.status == "EXPIRED":
raise ValueError("license.expired")
# Первая активация — запуск отсчёта срока
if license_obj.status == "UNACTIVATED":
now = _utcnow()
license_obj.activated_at = now
license_obj.expires_at = now + timedelta(days=license_obj.duration_days)
license_obj.status = "ACTIVE"
await db.flush()
# Проверка лимита устройств
existing_device_result = await db.execute(
select(Device).where(
Device.license_id == license_obj.id,
Device.device_id == device_id,
)
)
existing_device = existing_device_result.scalar_one_or_none()
if existing_device is None:
# Считаем активные устройства
active_count_result = await db.execute(
select(func.count(Device.id)).where(
Device.license_id == license_obj.id,
Device.is_active == True, # noqa: E712
)
)
active_count = active_count_result.scalar() or 0
if active_count >= license_obj.max_devices:
raise ValueError("device.limit_reached")
new_device = Device(
license_id=license_obj.id,
device_id=device_id,
platform=platform,
device_name=device_name,
is_active=True,
activated_at=_utcnow(),
last_heartbeat_at=_utcnow(),
)
db.add(new_device)
device_obj = new_device
else:
# Устройство уже привязано — обновляем heartbeat
existing_device.last_heartbeat_at = _utcnow()
existing_device.is_active = True
device_obj = existing_device
# Увеличиваем счётчик использований инвайта
if invite_obj is not None:
invite_obj.uses_count += 1
if invite_obj.uses_count >= invite_obj.max_uses:
invite_obj.is_active = False
await db.commit()
await db.refresh(license_obj)
await db.refresh(device_obj)
return license_obj, device_obj
async def heartbeat(
db: AsyncSession,
license_key: str,
device_id: str,
) -> dict:
"""Heartbeat: обновить last_heartbeat_at, вернуть статус подписки."""
license_obj = await get_license_by_raw_key(db, license_key)
if license_obj is None:
raise ValueError("license.not_found")
if license_obj.status == "REVOKED":
raise ValueError("license.revoked")
if license_obj.status == "MERGED":
raise ValueError("license.merged")
if license_obj.status == "UNACTIVATED":
raise ValueError("license.unactivated")
if license_obj.status == "EXPIRED":
raise ValueError("license.expired")
# Проверка срока
if license_obj.expires_at and license_obj.expires_at < _utcnow():
license_obj.status = "EXPIRED"
await db.commit()
raise ValueError("license.expired")
# Обновляем heartbeat устройства
device_result = await db.execute(
select(Device).where(
Device.license_id == license_obj.id,
Device.device_id == device_id,
)
)
device_obj = device_result.scalar_one_or_none()
if device_obj is not None:
device_obj.last_heartbeat_at = _utcnow()
await db.commit()
# Считаем дни до истечения
days_left = None
if license_obj.expires_at:
delta = license_obj.expires_at - _utcnow()
days_left = max(0, delta.days)
return {
"status": license_obj.status,
"expires_at": license_obj.expires_at,
"days_left": days_left,
}
async def unbind_device(
db: AsyncSession,
license_key: str,
device_id: str,
) -> bool:
"""Отвязать устройство от лицензии."""
license_obj = await get_license_by_raw_key(db, license_key)
if license_obj is None:
raise ValueError("license.not_found")
device_result = await db.execute(
select(Device).where(
Device.license_id == license_obj.id,
Device.device_id == device_id,
)
)
device_obj = device_result.scalar_one_or_none()
if device_obj is None:
raise ValueError("device.not_found")
await db.delete(device_obj)
await db.commit()
return True
async def list_devices(
db: AsyncSession,
license_obj: License,
) -> list[Device]:
"""Список устройств лицензии."""
result = await db.execute(
select(Device).where(Device.license_id == license_obj.id)
)
return list(result.scalars().all())
async def stack_key(
db: AsyncSession,
current_key: str,
new_key: str,
) -> tuple[License, KeyMergeHistory, bool, bool]:
"""Наложение/объединение сроков (Key Stacking).
- Текущий ключ должен быть ACTIVE и не истёк.
- Новый ключ должен быть UNACTIVATED.
- Прибавляет duration_days нового ключа к expires_at текущего.
- Если новый ключ выше по тарифу или больше слотов — авто-апгрейд.
- Новый ключ переводится в MERGED, пишется история.
"""
current = await get_license_by_raw_key(db, current_key)
if current is None:
raise ValueError("license.not_found")
if current.status != "ACTIVE":
raise ValueError("stack.current_not_active")
if current.expires_at and current.expires_at < _utcnow():
current.status = "EXPIRED"
await db.commit()
raise ValueError("stack.current_not_active")
new = await get_license_by_raw_key(db, new_key)
if new is None:
raise ValueError("license.not_found")
if new.status != "UNACTIVATED":
raise ValueError("stack.new_not_unactivated")
old_expires_at = current.expires_at
added_days = new.duration_days
new_expires_at = old_expires_at + timedelta(days=added_days)
old_edition = current.edition
old_max_devices = current.max_devices
# Апгрейд тарифа (если новый выше)
edition_upgraded = False
if EDITION_RANK[new.edition] > EDITION_RANK[current.edition]:
current.edition = new.edition
edition_upgraded = True
# Апгрейд слотов (если новый больше)
devices_upgraded = False
if new.max_devices > current.max_devices:
current.max_devices = new.max_devices
devices_upgraded = True
# Обновляем срок
current.expires_at = new_expires_at
current.status = "ACTIVE"
# Новый ключ → MERGED
new.status = "MERGED"
# Запись в историю
history = KeyMergeHistory(
target_license_id=current.id,
source_license_id=new.id,
source_key_masked=new.key_masked,
old_expires_at=old_expires_at,
new_expires_at=new_expires_at,
added_days=added_days,
old_edition=old_edition,
new_edition=current.edition,
old_max_devices=old_max_devices,
new_max_devices=current.max_devices,
)
db.add(history)
await db.commit()
await db.refresh(current)
await db.refresh(history)
return current, history, edition_upgraded, devices_upgraded
async def switch_key(
db: AsyncSession,
current_key: str,
new_key: str,
) -> tuple[License, KeyMergeHistory, str]:
"""Двустороннее переключение тарифа «на лету» (Upgrade & Downgrade).
- Текущий ключ ACTIVE и не истёк; новый UNACTIVATED.
- Тариф ВСЕГДА переходит на edition нового ключа (вверх или вниз).
- Срок нового ключа прибавляется к текущему expires_at.
- max_devices становится как у нового ключа.
- Новый ключ → MERGED, пишется история.
- При понижении ниже TOTAL_SECURITY облачный Vault → read_only
(вычисляется на уровне API по рангу тарифа, здесь не трогаем данные).
"""
current = await get_license_by_raw_key(db, current_key)
if current is None:
raise ValueError("license.not_found")
if current.status != "ACTIVE":
raise ValueError("stack.current_not_active")
if current.expires_at and current.expires_at < _utcnow():
current.status = "EXPIRED"
await db.commit()
raise ValueError("stack.current_not_active")
new = await get_license_by_raw_key(db, new_key)
if new is None:
raise ValueError("license.not_found")
if new.status != "UNACTIVATED":
raise ValueError("stack.new_not_unactivated")
old_expires_at = current.expires_at
added_days = new.duration_days
new_expires_at = old_expires_at + timedelta(days=added_days)
old_edition = current.edition
old_max_devices = current.max_devices
direction = "upgrade"
if EDITION_RANK.get(new.edition, 0) < EDITION_RANK.get(current.edition, 0):
direction = "downgrade"
# Всегда переключаем на тариф нового ключа
current.edition = new.edition
current.max_devices = new.max_devices
current.expires_at = new_expires_at
current.status = "ACTIVE"
# Новый ключ → MERGED
new.status = "MERGED"
history = KeyMergeHistory(
target_license_id=current.id,
source_license_id=new.id,
source_key_masked=new.key_masked,
old_expires_at=old_expires_at,
new_expires_at=new_expires_at,
added_days=added_days,
old_edition=old_edition,
new_edition=current.edition,
old_max_devices=old_max_devices,
new_max_devices=current.max_devices,
)
db.add(history)
await db.commit()
await db.refresh(current)
await db.refresh(history)
return current, history, direction
async def create_invite(
db: AsyncSession,
license_key: str,
max_uses: int,
) -> tuple[Invite, str]:
"""Создать одноразовый инвайт-код для лицензии."""
license_obj = await get_license_by_raw_key(db, license_key)
if license_obj is None:
raise ValueError("license.not_found")
raw_token = generate_invite_token()
invite_obj = Invite(
license_id=license_obj.id,
token_hash=hash_key(raw_token),
token_masked=mask_token(raw_token),
max_uses=max_uses,
uses_count=0,
is_active=True,
region="GLOBAL",
)
db.add(invite_obj)
await db.commit()
await db.refresh(invite_obj)
return invite_obj, raw_token
async def list_invites(
db: AsyncSession,
license_obj: License,
) -> list[Invite]:
"""Список инвайтов лицензии."""
result = await db.execute(
select(Invite).where(Invite.license_id == license_obj.id)
)
return list(result.scalars().all())
======================================================================МОДУЛЬ: Сервис восстановленияФАЙЛ: app/services/recovery_service.py======================================================================
# -*- coding: utf-8 -*-
"""Сервис восстановления ключа без Email (Zero-Telemetry Recovery).
Хранит ТОЛЬКО SHA-256 с солью — сам код/фраза не сохраняются на сервере.
Rate-limit перебора: Redis (5 попыток / 10 мин на клиентский ключ).
Способы восстановления:
1. По HWID (auto): устройство уже привязано → вернуть профиль.
2. По числу XXX-XXX-XX (генератор в 1 клик).
3. По связке secret_word + pin_digits (своя фраза).
"""
import hashlib
import random
import re
import time
SALT = "cifra_secure_recovery_v2:"
RATE_LIMIT_ATTEMPTS = 5
RATE_LIMIT_WINDOW = 600 # 10 минут
def hash_recovery(value: str) -> str:
"""SHA-256 с солью от нормализованного значения."""
normalized = value.strip().lower()
return hashlib.sha256((SALT + normalized).encode("utf-8")).hexdigest()
def generate_numeric_code() -> str:
"""Короткий уникальный цифровой код XXX-XXX-XX (например 742-819-05)."""
a = random.randint(0, 999)
b = random.randint(0, 999)
c = random.randint(0, 99)
return f"{a:03d}-{b:03d}-{c:02d}"
def normalize_code(raw: str) -> str:
"""Нормализовать введённый код к формату XXX-XXX-XX (только цифры → с дефисами)."""
digits = re.sub(r"\D", "", raw)
if len(digits) == 8:
return f"{digits[0:3]}-{digits[3:6]}-{digits[6:8]}"
return raw.strip()
def build_phrase_hash(secret_word: str, pin_digits: str) -> str:
"""Хэш связки слово + 4 цифры."""
return hash_recovery(f"{secret_word.strip().lower()}|{pin_digits.strip()}")
# ── Rate-limit (Redis) ────────────────────────────────────────
async def check_rate_limit(redis_client, client_id: str) -> bool:
"""True если можно пытаться (лимит не исчерпан), False если заблокирован."""
key = f"recovery_rl:{client_id}"
attempts = await redis_client.get(key)
if attempts is None:
return True
try:
return int(attempts) < RATE_LIMIT_ATTEMPTS
except (TypeError, ValueError):
return True
async def register_attempt(redis_client, client_id: str):
"""Зафиксировать попытку. Устанавливает окно 10 мин при первой попытке."""
key = f"recovery_rl:{client_id}"
current = await redis_client.incr(key)
if current == 1:
await redis_client.expire(key, RATE_LIMIT_WINDOW)
async def remaining_attempts(redis_client, client_id: str) -> int:
"""Сколько попыток осталось."""
key = f"recovery_rl:{client_id}"
attempts = await redis_client.get(key)
if attempts is None:
return RATE_LIMIT_ATTEMPTS
try:
return max(0, RATE_LIMIT_ATTEMPTS - int(attempts))
except (TypeError, ValueError):
return RATE_LIMIT_ATTEMPTS
======================================================================МОДУЛЬ: Сервис телефоновФАЙЛ: app/services/phone_service.py======================================================================
# -*- coding: utf-8 -*-
"""Сервис спам-номеров (Who Calls): нормализация + облачный лукап."""
import re
import json
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from redis.asyncio import Redis
from app.models import SpamPhones
def normalize_phone(raw: str) -> str:
"""Нормализовать номер: оставить только цифры, привести к международному виду.
- Убирает всё кроме цифр и ведущего '+'.
- RU-номера 8XXXXXXXXXX → +7XXXXXXXXXX.
- RU-номера 7XXXXXXXXXX → +7XXXXXXXXXX.
- Короткие (без кода) → как есть (цифры).
"""
digits = re.sub(r"\D", "", raw)
if not digits:
return ""
# Российские мобильные: 8XXXXXXXXXX (11 цифр, начинается с 8) → +7...
if len(digits) == 11 and digits.startswith("8"):
return "+7" + digits[1:]
# Российские: 7XXXXXXXXXX (11 цифр, начинается с 7) → +7...
if len(digits) == 11 and digits.startswith("7"):
return "+" + digits
# Прочие международные: если начинается с 00 → +...
if raw.strip().startswith("00"):
return "+" + digits
# Иначе — просто цифры с ведущим +, если есть
return "+" + digits if raw.strip().startswith("+") else digits
async def lookup_phone(
db: AsyncSession,
redis_client: Redis,
raw_phone: str,
) -> dict:
"""Проверить номер. Возвращает {phone, status, category, description_ru, description_en}.
Сначала Redis-кэш, затем таблица spam_phones.
"""
normalized = normalize_phone(raw_phone)
if not normalized:
return {"phone": raw_phone, "status": "CLEAN", "category": None,
"description_ru": None, "description_en": None}
# Redis-кэш
cache_key = f"phone:{normalized}"
cached = await redis_client.get(cache_key)
if cached:
return json.loads(cached)
# Таблица spam_phones
result = await db.execute(
select(SpamPhones).where(SpamPhones.phone_normalized == normalized)
)
row = result.scalar_one_or_none()
if row is None:
entry = {"phone": normalized, "status": "CLEAN", "category": None,
"description_ru": None, "description_en": None}
else:
status = "CLEAN" if row.category == "SAFE" else "SPAM"
entry = {
"phone": normalized,
"status": status,
"category": row.category,
"description_ru": row.description_ru,
"description_en": row.description_en,
}
# Кэшируем на 1 час
await redis_client.setex(cache_key, 3600, json.dumps(entry, ensure_ascii=False))
return entry
async def add_phone(
db: AsyncSession,
redis_client: Redis,
raw_phone: str,
category: str,
description_ru: str | None,
description_en: str | None,
) -> tuple[SpamPhones, bool]:
"""Добавить номер в базу. Возвращает (объект, created)."""
normalized = normalize_phone(raw_phone)
existing = await db.execute(
select(SpamPhones).where(SpamPhones.phone_normalized == normalized)
)
found = existing.scalar_one_or_none()
if found is not None:
return found, False
row = SpamPhones(
phone_normalized=normalized,
category=category,
description_ru=description_ru,
description_en=description_en,
)
db.add(row)
await db.commit()
await db.refresh(row)
# Прогреваем Redis
entry = {
"phone": normalized,
"status": "CLEAN" if category == "SAFE" else "SPAM",
"category": category,
"description_ru": description_ru,
"description_en": description_en,
}
await redis_client.setex(f"phone:{normalized}", 3600, json.dumps(entry, ensure_ascii=False))
return row, True
======================================================================МОДУЛЬ: Синхронизация угрозФАЙЛ: app/feed_updater.py======================================================================
# -*- coding: utf-8 -*-
"""Авто-обновление вирусных баз из MalwareBazaar (Abuse.ch Threat Feed Updater).
Скачивает свежий список SHA-256 реальных угроз из публичного фида
Abuse.ch MalwareBazaar и массово добавляет их в threat_signatures + Redis
со статусом MALWARE (threat_name = MalwareBazaar.RecentThreat).
Источники (по приоритету):
1. https://bazaar.abuse.ch/export/txt/recent/ (txt-экспорт последних сигнатур)
2. https://mb-api.abuse.ch/api/v1/ (POST query=get_recent, если txt недоступен)
3. Fallback: локальный кэш-файл /tmp/mb_cache.txt (последний удачный скачанный фид)
4. Fallback: генератор демо-хэшей (если внешний интернет в песочнице полностью ограничен)
Использование (внутри контейнера api):
python -m app.feed_updater # одиночный синк
Или через эндпоинт:
POST /api/v1/admin/signatures/sync-feed
"""
import asyncio
import hashlib
import json
import os
import re
import urllib.request
import urllib.parse
from datetime import datetime, timezone
from sqlalchemy import select
from sqlalchemy.dialects.postgresql import insert as pg_insert
from app.database import AsyncSessionLocal, redis_client, init_db
from app.models import ThreatSignatures
THREAT_NAME = "MalwareBazaar.RecentThreat"
SEVERITY = "HIGH"
CACHE_FILE = "/tmp/mb_cache.txt"
TIMEOUT = 20
# SHA-256: 64 hex-символа
_SHA256_RE = re.compile(r"\b[a-fA-F0-9]{64}\b")
def _http_get(url: str, timeout: int = TIMEOUT) -> str | None:
"""GET-запрос с UA. Возвращает text или None при ошибке."""
try:
req = urllib.request.Request(url, headers={"User-Agent": "AV-Platform-FeedUpdater/1.0"})
with urllib.request.urlopen(req, timeout=timeout) as r:
return r.read().decode("utf-8", errors="ignore")
except Exception:
return None
def _http_post_json(url: str, data: dict, timeout: int = TIMEOUT) -> dict | None:
"""POST JSON-запрос. Возвращает dict или None."""
try:
body = urllib.parse.urlencode(data).encode()
req = urllib.request.Request(
url, data=body,
headers={"User-Agent": "AV-Platform-FeedUpdater/1.0",
"Content-Type": "application/x-www-form-urlencoded"},
)
with urllib.request.urlopen(req, timeout=timeout) as r:
return json.loads(r.read().decode("utf-8", errors="ignore"))
except Exception:
return None
def parse_hashes_from_text(text: str) -> list[str]:
"""Извлечь уникальные SHA-256 из txt/JSON-текста фида."""
seen: dict[str, None] = {}
for m in _SHA256_RE.finditer(text):
h = m.group(0).lower()
seen.setdefault(h)
return list(seen.keys())
def fetch_recent_hashes() -> tuple[list[str], str]:
"""Скачать свежие SHA-256. Возвращает (hashes, source).
Порядок: txt-экспорт → API → кэш-файл → генератор.
"""
# 1) txt-экспорт MalwareBazaar (последние сигнатуры)
text = _http_get("https://bazaar.abuse.ch/export/txt/recent/")
if text and len(text) > 200:
hashes = parse_hashes_from_text(text)
if len(hashes) >= 10:
_save_cache(text)
return hashes, "bazaar.abuse.ch/export/txt/recent"
# 2) API mb-api.abuse.ch (get_recent)
resp = _http_post_json("https://mb-api.abuse.ch/api/v1/", {"query": "get_recent"})
if resp and resp.get("query_status") == "ok":
data = resp.get("data", [])
hashes = []
for item in data:
sha = item.get("sha256_hash") or item.get("sha256")
if sha and _SHA256_RE.fullmatch(sha):
hashes.append(sha.lower())
if len(hashes) >= 10:
_save_cache(json.dumps({"hashes": hashes}))
return hashes, "mb-api.abuse.ch/api/v1/get_recent"
# 3) локальный кэш-файл (последний удачный фид)
cached = _load_cache()
if cached and len(cached) >= 10:
return cached, "local-cache"
# 4) генератор демо-хэшей (внешний интернет полностью недоступен)
demo = _generate_demo_hashes(50)
return demo, "demo-generator"
def _save_cache(text: str) -> None:
try:
with open(CACHE_FILE, "w") as f:
f.write(text)
except Exception:
pass
def _load_cache() -> list[str]:
try:
with open(CACHE_FILE) as f:
text = f.read()
# JSON-формат {"hashes": [...]} или txt
try:
obj = json.loads(text)
if isinstance(obj, dict) and "hashes" in obj:
return [h.lower() for h in obj["hashes"] if _SHA256_RE.fullmatch(h)]
except json.JSONDecodeError:
pass
return parse_hashes_from_text(text)
except Exception:
return []
def _generate_demo_hashes(n: int) -> list[str]:
"""Детерминированные демо-хэши (для оффлайн-песочницы)."""
out = []
for i in range(n):
out.append(hashlib.sha256(f"malwarebazaar-demo-{i}".encode()).hexdigest())
return out
async def sync_feed() -> dict:
"""Скачать фид и массово внести в threat_signatures + Redis.
Возвращает статистику {source, fetched, added, skipped}.
"""
await init_db()
hashes, source = fetch_recent_hashes()
fetched = len(hashes)
added = 0
skipped = 0
now = datetime.now(timezone.utc)
async with AsyncSessionLocal() as session:
# Уже существующие хэши (чтобы не дублировать)
existing = await session.execute(
select(ThreatSignatures.sha256).where(ThreatSignatures.sha256.in_(hashes))
)
existing_set = {r[0] for r in existing.all()}
new_rows = []
for h in hashes:
if h in existing_set:
skipped += 1
continue
new_rows.append(ThreatSignatures(
sha256=h,
threat_name=THREAT_NAME,
severity=SEVERITY,
family="MalwareBazaar",
created_at=now,
))
# Прогреваем Redis-кэш
entry = {"sha256": h, "status": "MALWARE", "threat_name": THREAT_NAME}
await redis_client.setex(f"threat:{h}", 3600, json.dumps(entry))
added += 1
session.add_all(new_rows)
await session.commit()
return {
"source": source,
"fetched": fetched,
"added": added,
"skipped": skipped,
"threat_name": THREAT_NAME,
"message": f"Синк завершён: источник={source}, скачано={fetched}, добавлено={added}, дублей={skipped}.",
}
if __name__ == "__main__":
result = asyncio.run(sync_feed())
print(json.dumps(result, ensure_ascii=False, indent=2))
======================================================================МОДУЛЬ: Disaster RecoveryФАЙЛ: app/backup_mirror.py======================================================================
# -*- coding: utf-8 -*-
"""CIFRA SECURE — Disaster Recovery: экспорт/импорт БД для Geo-Failover зеркал.
Снимок (snapshot) всех бизнес-таблиц в JSON. Импорт выполняет UPSERT
(ON CONFLICT ... DO UPDATE) — 100% ключей/устройств сохраняется при
синхронизации Primary → Mirror (KZ/EU) или восстановлении после сбоя.
"""
from datetime import datetime, date
from sqlalchemy import select, inspect
from sqlalchemy.dialects.postgresql import insert as pg_insert
from sqlalchemy.ext.asyncio import AsyncSession
from app.models import (
License, Device, Invite, PasswordVault, ThreatSignatures,
SpamPhones, KeyMergeHistory,
)
# Таблицы, попадающие в слепок (в порядке зависимостей: сначала родители)
TABLES = [
(License, "licenses"),
(Device, "devices"),
(Invite, "invites"),
(PasswordVault, "password_vaults"),
(ThreatSignatures, "threat_signatures"),
(SpamPhones, "spam_phones"),
(KeyMergeHistory, "key_merge_history"),
]
def _serialize_row(row) -> dict:
"""Преобразовать ORM-строку в JSON-совместимый dict."""
out = {}
for col in inspect(type(row)).columns:
val = getattr(row, col.name)
if isinstance(val, (datetime, date)):
val = val.isoformat()
out[col.name] = val
return out
async def export_snapshot(db: AsyncSession) -> dict:
"""Собрать полный слепок всех таблиц."""
snap = {
"meta": {
"product": "CIFRA SECURE",
"exported_at": datetime.utcnow().isoformat(),
"version": 1,
},
}
counts = {}
for model, name in TABLES:
result = await db.execute(select(model))
rows = result.scalars().all()
snap[name] = [_serialize_row(r) for r in rows]
counts[name] = len(rows)
snap["meta"]["counts"] = counts
return snap
def _parse_dt(val):
"""Преобразовать ISO-строку обратно в datetime (для колонок-дат)."""
if isinstance(val, str):
try:
return datetime.fromisoformat(val)
except ValueError:
return val
return val
async def import_snapshot(db: AsyncSession, snap: dict) -> dict:
"""Импортировать слепок (UPSERT по primary key). Возвращает статистику."""
imported = {}
for model, name in TABLES:
rows = snap.get(name, [])
if not rows:
imported[name] = 0
continue
# Фильтруем только существующие колонки (устойчивость к версиям)
model_cols = {c.name for c in inspect(model).columns}
clean_rows = [{k: v for k, v in r.items() if k in model_cols} for r in rows]
clean_rows = [r for r in clean_rows if r]
# Восстанавливаем datetime из ISO-строк
date_cols = {c.name for c in inspect(model).columns
if str(c.type).upper().startswith(("TIMESTAMP", "DATETIME", "DATE"))}
for r in clean_rows:
for dc in date_cols:
if dc in r:
r[dc] = _parse_dt(r[dc])
if not clean_rows:
imported[name] = 0
continue
# Определяем conflict target (первичный ключ / unique)
pk_cols = [c.name for c in inspect(model).primary_key]
if not pk_cols:
pk_cols = list(clean_rows[0].keys())[:1]
stmt = pg_insert(model).values(clean_rows)
update_cols = {c: stmt.excluded[c] for c in clean_rows[0].keys()
if c not in pk_cols}
if update_cols:
stmt = stmt.on_conflict_do_update(index_elements=pk_cols, set_=update_cols)
else:
stmt = stmt.on_conflict_do_nothing(index_elements=pk_cols)
await db.execute(stmt)
imported[name] = len(clean_rows)
await db.commit()
# Верификация: 100% записей на месте
verify = await export_snapshot(db)
return {"imported": imported, "verified_counts": verify["meta"]["counts"]}
======================================================================МОДУЛЬ: Сканер (движок)ФАЙЛ: windows-agent/core/scanner.py======================================================================
# -*- coding: utf-8 -*-
"""Сканер: двойной движок проверки файлов.
1) Облако: POST /api/v1/cloud/lookup по SHA-256.
2) Локальный офлайн: EICAR + 4 эвристики + In-Memory ZIP-сканирование.
"""
import hashlib
import json
import urllib.request
import zipfile
from pathlib import Path
from core.config import API_BASE, MAX_SCAN_SIZE_MB
# EICAR Standard Antivirus Test File (SHA-256)
EICAR_SHA256 = "275a021bbfb6489e54d471899f7db9d1663fc695ec2fe2a2c4538aabf651fd0f"
# EICAR строка (для локальной проверки содержимого)
EICAR_STRING = b"X5O!P%@AP[4\\PZX54(P^)7CC)7}$EICAR-STANDARD-ANTIVIRUS-TEST-FILE!$H+H*"
# ── Оффлайн-эвристики (4 правила, без интернета) ─────────────
# (threat_name, [маркеры-последовательности, все должны найтись в буфере])
HEURISTICS = [
("Heuristic.CoinMiner.Stratum", [b"stratum+tcp://", b"xmrig", b"donate-level"]),
("Heuristic.Stealer.CredHarvest", [b"Login Data", b"tdata", b"Local State"]),
("Heuristic.Ransomware.Note", [b"vssadmin delete shadows", b"YOUR FILES ARE ENCRYPTED"]),
]
# Защита от ZIP-бомб
ZIP_MAX_UNCOMPRESSED_MB = 50
ZIP_MAX_DEPTH = 3
ZIP_MAX_ENTRIES = 500
def sha256_file(path: Path) -> str | None:
"""Вычислить SHA-256 файла."""
h = hashlib.sha256()
try:
with open(path, "rb") as f:
for chunk in iter(lambda: f.read(1024 * 1024), b""):
h.update(chunk)
return h.hexdigest()
except OSError:
return None
def _check_heuristics(buffer: bytes) -> tuple[str, str | None]:
"""Проверить буфер по 4 эвристическим правилам (все маркеры должны найтись)."""
lower = buffer.lower()
for name, markers in HEURISTICS:
if all(m.lower() in lower for m in markers):
return "MALWARE", name
return "UNKNOWN", None
def local_check(path: Path, file_hash: str) -> tuple[str, str | None]:
"""Локальная офлайн-проверка (EICAR + эвристики + ZIP).
Возвращает (status, threat_name).
"""
# EICAR по хэшу
if file_hash == EICAR_SHA256:
return "MALWARE", "EICAR-Test-Signature"
# Читаем содержимое (до 256 КБ достаточно для маркеров)
try:
with open(path, "rb") as f:
data = f.read(256 * 1024)
except OSError:
data = b""
# EICAR по содержимому
if EICAR_STRING in data:
return "MALWARE", "EICAR-Test-Signature"
# 4 эвристики
st, threat = _check_heuristics(data)
if st == "MALWARE":
return st, threat
# In-Memory ZIP-сканирование (EICAR/эвристики внутри архива)
if zipfile.is_zipfile(path):
return scan_zip_contents(path)
return "UNKNOWN", None
def scan_zip_contents(zip_path: Path, depth: int = 0) -> tuple[str, str | None]:
"""In-Memory сканирование ZIP (с защитой от бомб и вложенности).
Проверяет каждый файл внутри архива на EICAR + эвристики.
Поддерживает вложенные .zip (до ZIP_MAX_DEPTH).
"""
if depth > ZIP_MAX_DEPTH:
return "UNKNOWN", None
try:
with zipfile.ZipFile(zip_path) as zf:
total_uncompressed = 0
entries = 0
for info in zf.infolist():
entries += 1
if entries > ZIP_MAX_ENTRIES:
break # защита от бомбы по числу файлов
# Защита по суммарному распакованному размеру
total_uncompressed += info.file_size
if total_uncompressed > ZIP_MAX_UNCOMPRESSED_MB * 1024 * 1024:
return "UNKNOWN", None # бомба — пропускаем, не распаковываем
if info.is_dir():
continue
try:
# Читаем в память (не распаковывая на диск)
data = zf.read(info.filename)
except Exception:
continue
# EICAR внутри zip
if EICAR_STRING in data[:256 * 1024]:
return "MALWARE", "EICAR-Test-Signature (in ZIP)"
# Эвристики внутри zip
st, threat = _check_heuristics(data[:256 * 1024])
if st == "MALWARE":
return st, f"{threat} (in ZIP)"
# Вложенный .zip → рекурсия
if info.filename.lower().endswith(".zip"):
try:
import io
with zipfile.ZipFile(io.BytesIO(data)) as nested:
pass # валидность проверена
# Рекурсивно сканируем вложенный архив через временный подход
st2, threat2 = _scan_nested_zip_bytes(data, depth + 1)
if st2 == "MALWARE":
return st2, threat2
except Exception:
pass
except zipfile.BadZipFile:
return "UNKNOWN", None
except Exception:
return "UNKNOWN", None
return "UNKNOWN", None
def _scan_nested_zip_bytes(data: bytes, depth: int) -> tuple[str, str | None]:
"""Сканировать вложенный ZIP из байтов в памяти."""
import io
if depth > ZIP_MAX_DEPTH:
return "UNKNOWN", None
try:
with zipfile.ZipFile(io.BytesIO(data)) as zf:
for info in zf.infolist():
if info.is_dir():
continue
try:
inner = zf.read(info.filename)
except Exception:
continue
if EICAR_STRING in inner[:256 * 1024]:
return "MALWARE", "EICAR-Test-Signature (nested ZIP)"
st, threat = _check_heuristics(inner[:256 * 1024])
if st == "MALWARE":
return st, f"{threat} (nested ZIP)"
if info.filename.lower().endswith(".zip"):
st2, threat2 = _scan_nested_zip_bytes(inner, depth + 1)
if st2 == "MALWARE":
return st2, threat2
except Exception:
pass
return "UNKNOWN", None
def cloud_lookup(file_hash: str, license_key: str | None = None) -> tuple[str, str | None]:
"""Запрос к облачному API. Возвращает (status, threat_name)."""
url = f"{API_BASE}/cloud/lookup"
payload = json.dumps({"hashes": [file_hash]}).encode("utf-8")
headers = {"Content-Type": "application/json"}
if license_key:
headers["X-License-Key"] = license_key
req = urllib.request.Request(url, data=payload, headers=headers, method="POST")
try:
with urllib.request.urlopen(req, timeout=10) as resp:
data = json.loads(resp.read().decode("utf-8"))
results = data.get("results", [])
if results:
r = results[0]
return r.get("status", "UNKNOWN"), r.get("threat_name")
except Exception:
pass
return "UNKNOWN", None
def scan_file(path: Path, license_key: str | None = None, cache=None) -> dict:
"""Полная проверка файла (облако + локально).
Если cache передан и файл в белом списке (Exclusions) — мгновенный
CLEAN (WHITELISTED) за 0 мс (без хэширования/сети).
Возвращает dict: {path, sha256, status, threat_name, engine}.
"""
# 0) Белый список (Exclusions) — самый быстрый путь, до всего остального
if cache is not None and cache.is_whitelisted(path):
return {"path": str(path), "status": "CLEAN", "threat_name": None,
"engine": "whitelist", "whitelisted": True}
# Проверка размера
try:
size_mb = path.stat().st_size / (1024 * 1024)
if size_mb > MAX_SCAN_SIZE_MB:
return {"path": str(path), "status": "SKIPPED", "threat_name": None, "engine": "size-limit"}
except OSError:
return {"path": str(path), "status": "ERROR", "threat_name": None, "engine": "fs"}
file_hash = sha256_file(path)
if file_hash is None:
return {"path": str(path), "status": "ERROR", "threat_name": None, "engine": "fs"}
# 0b) Белый список по sha256
if cache is not None and cache.is_whitelisted(path, file_hash):
return {"path": str(path), "sha256": file_hash, "status": "CLEAN",
"threat_name": None, "engine": "whitelist", "whitelisted": True}
# 1) Локальная проверка (быстрая)
local_status, local_threat = local_check(path, file_hash)
if local_status == "MALWARE":
return {"path": str(path), "sha256": file_hash, "status": "MALWARE", "threat_name": local_threat, "engine": "local"}
# 2) Облачный лукап
cloud_status, cloud_threat = cloud_lookup(file_hash, license_key)
if cloud_status == "MALWARE":
return {"path": str(path), "sha256": file_hash, "status": "MALWARE", "threat_name": cloud_threat, "engine": "cloud"}
# Если облако дало определённый статус CLEAN — используем его
if cloud_status == "CLEAN":
return {"path": str(path), "sha256": file_hash, "status": "CLEAN", "threat_name": None, "engine": "cloud"}
return {"path": str(path), "sha256": file_hash, "status": "UNKNOWN", "threat_name": None, "engine": "combined"}
======================================================================МОДУЛЬ: Мониторинг файловФАЙЛ: windows-agent/core/monitor.py======================================================================
# -*- coding: utf-8 -*-
"""Мониторинг файлов: фоновый опрос критических папок.
Реализация на polling (без внешних зависимостей вроде watchdog).
Комбинируется со Smart Cache: уже проверенные файлы (size+mtime) пропускаются мгновенно.
"""
import threading
import time
from pathlib import Path
from core.config import critical_folders
from core.cache import ScanCache
from core.scanner import scan_file
from core.quarantine import quarantine_file
class FileMonitor:
"""Фоновый мониторинг критических папок."""
def __init__(self, cache: ScanCache, license_key: str | None = None,
on_threat=None, on_scan=None):
self.cache = cache
self.license_key = license_key
self.on_threat = on_threat # callback(path, result)
self.on_scan = on_scan # callback(path, result)
self._stop = threading.Event()
self._thread = None
self.scanned_count = 0
self.threat_count = 0
self._seen: set[str] = set()
# Таймер паузы защиты (ESET-style): epoch-время окончания паузы
self.pause_until: float | None = None # None = защита активна
# ── Пауза защиты (Pause Protection) ────────────────────────
def pause(self, seconds: float | None = None):
"""Поставить защиту на паузу.
seconds=None → «До перезагрузки» (бесконечно, пока не resume()).
"""
if seconds is None:
self.pause_until = float("inf")
else:
self.pause_until = time.time() + seconds
def resume(self):
"""Возобновить защиту (сбросить паузу)."""
self.pause_until = None
def is_paused(self) -> bool:
"""Активна ли сейчас пауза."""
if self.pause_until is None:
return False
if self.pause_until == float("inf"):
return True
if time.time() >= self.pause_until:
# Пауза истекла → авто-возобновление
self.pause_until = None
return False
return True
def pause_remaining(self) -> float | None:
"""Секунды до конца паузы (None = не на паузе, inf = до перезагрузки)."""
if not self.is_paused():
return None
if self.pause_until == float("inf"):
return float("inf")
return max(0.0, self.pause_until - time.time())
def start(self):
"""Запустить фоновый мониторинг."""
if self._thread and self._thread.is_alive():
return
self._stop.clear()
self._thread = threading.Thread(target=self._loop, daemon=True)
self._thread.start()
def stop(self):
"""Остановить мониторинг."""
self._stop.set()
def _loop(self):
"""Основной цикл опроса."""
while not self._stop.is_set():
# Если защита на паузе — пропускаем проверки (таймер сам возобновит)
if not self.is_paused():
try:
self._scan_once()
except Exception:
pass
# Пауза между опросами
self._stop.wait(timeout=10)
def _scan_once(self):
"""Один проход по критическим папкам."""
for folder in critical_folders():
if self._stop.is_set():
return
self._scan_folder(folder)
def _scan_folder(self, folder: Path, depth: int = 0):
"""Рекурсивный обход папки (ограниченная глубина)."""
if depth > 4 or self._stop.is_set():
return
try:
entries = list(folder.iterdir())
except PermissionError:
return
for entry in entries:
if self._stop.is_set():
return
try:
if entry.is_dir():
self._scan_folder(entry, depth + 1)
elif entry.is_file():
self._check_file(entry)
except (PermissionError, OSError):
continue
def _check_file(self, path: Path):
"""Проверить один файл (с кэшем)."""
key = str(path)
if key in self._seen:
return
# Smart Cache: если size+mtime не изменились — пропуск
cached = self.cache.get(path)
if cached is not None:
self._seen.add(key)
status = cached["status"]
if status == "MALWARE":
self.threat_count += 1
if self.on_threat:
self.on_threat(path, {"status": status, "threat_name": cached["threat_name"]})
return
# Полная проверка (cache → whitelist-проверка внутри scan_file)
result = scan_file(path, self.license_key, cache=self.cache)
self.scanned_count += 1
self._seen.add(key)
if result["status"] in ("MALWARE", "CLEAN", "UNKNOWN"):
self.cache.put(
path,
result.get("sha256", ""),
result["status"],
result.get("threat_name"),
)
if result["status"] == "MALWARE":
self.threat_count += 1
# Авто-карантин
quarantine_file(path)
if self.on_threat:
self.on_threat(path, result)
if self.on_scan:
self.on_scan(path, result)
def stats(self) -> dict:
"""Статистика мониторинга."""
return {
"scanned": self.scanned_count,
"threats": self.threat_count,
"running": self._thread is not None and self._thread.is_alive(),
}
======================================================================МОДУЛЬ: КарантинФАЙЛ: windows-agent/core/quarantine.py======================================================================
# -*- coding: utf-8 -*-
"""Карантин: безопасная изоляция заражённых файлов.
Заражённый файл:
1) XOR-кодируется (блокирует прямое выполнение).
2) Перемещается в папку карантина.
3) Записывается метаданных для восстановления.
"""
import json
import os
import time
from pathlib import Path
from core.config import QUARANTINE_DIR
# XOR-ключ (obfuscation, не криптографическая защита —只是为了阻止 запуск)
XOR_KEY = b"\xA5\x3C\x7E\x19\xB2\x04\xD8\x61"
def _xor_encode(data: bytes) -> bytes:
key_len = len(XOR_KEY)
return bytes(b ^ XOR_KEY[i % key_len] for i, b in enumerate(data))
def quarantine_file(path: Path) -> dict:
"""Изолировать заражённый файл в карантин.
Возвращает dict с метаданными.
"""
if not path.exists():
return {"success": False, "error": "file_not_found"}
original_path = str(path)
original_name = path.name
qname = f"{int(time.time())}_{original_name}.quar"
qpath = QUARANTINE_DIR / qname
try:
# Читаем, кодируем, пишем в карантин
data = path.read_bytes()
encoded = _xor_encode(data)
qpath.write_bytes(encoded)
# Удаляем оригинал
path.unlink()
# Метаданные
meta = {
"original_path": original_path,
"original_name": original_name,
"quarantine_name": qname,
"size": len(data),
"quarantined_at": time.time(),
}
meta_path = QUARANTINE_DIR / f"{qname}.meta.json"
meta_path.write_text(json.dumps(meta, ensure_ascii=False, indent=2), encoding="utf-8")
return {"success": True, "quarantine_name": qname, "original_path": original_path}
except Exception as e:
return {"success": False, "error": str(e)}
def list_quarantine() -> list[dict]:
"""Список файлов в карантине."""
items = []
for meta_path in QUARANTINE_DIR.glob("*.meta.json"):
try:
meta = json.loads(meta_path.read_text(encoding="utf-8"))
items.append(meta)
except Exception:
continue
return sorted(items, key=lambda x: x.get("quarantined_at", 0), reverse=True)
def restore_file(quarantine_name: str) -> dict:
"""Восстановить файл из карантина (раскодировать + вернуть на место)."""
qpath = QUARANTINE_DIR / quarantine_name
meta_path = QUARANTINE_DIR / f"{quarantine_name}.meta.json"
if not qpath.exists() or not meta_path.exists():
return {"success": False, "error": "not_found"}
try:
meta = json.loads(meta_path.read_text(encoding="utf-8"))
encoded = qpath.read_bytes()
decoded = _xor_encode(encoded) # XOR обратим
original_path = Path(meta["original_path"])
original_path.parent.mkdir(parents=True, exist_ok=True)
original_path.write_bytes(decoded)
qpath.unlink()
meta_path.unlink()
return {"success": True, "restored_to": str(original_path)}
except Exception as e:
return {"success": False, "error": str(e)}
def delete_from_quarantine(quarantine_name: str) -> dict:
"""Окончательно удалить файл из карантина."""
qpath = QUARANTINE_DIR / quarantine_name
meta_path = QUARANTINE_DIR / f"{quarantine_name}.meta.json"
try:
if qpath.exists():
qpath.unlink()
if meta_path.exists():
meta_path.unlink()
return {"success": True}
except Exception as e:
return {"success": False, "error": str(e)}
======================================================================МОДУЛЬ: Failover-клиентФАЙЛ: windows-agent/core/license.py======================================================================
# -*- coding: utf-8 -*-
"""Лицензия: активация, heartbeat, хранение ключа локально.
Поддерживает Multi-Mirror Geo-Failover (API_ENDPOINTS) и Offline Grace Mode
(offline_license_cache.json): при недоступности всех зеркал агент продолжает
работать до OFFLINE_GRACE_DAYS с последней успешной проверкой.
"""
import json
import time
import urllib.request
from core.config import (
API_BASE, API_ENDPOINTS, KEY_FILE,
OFFLINE_CACHE_FILE, OFFLINE_GRACE_DAYS,
)
def save_key(key: str):
"""Сохранить лицензионный ключ локально."""
KEY_FILE.write_text(key.strip(), encoding="utf-8")
def load_key() -> str | None:
"""Загрузить сохранённый ключ."""
if KEY_FILE.exists():
k = KEY_FILE.read_text(encoding="utf-8").strip()
return k or None
return None
def clear_key():
"""Удалить сохранённый ключ (выход)."""
if KEY_FILE.exists():
KEY_FILE.unlink()
# ── Offline Grace Mode ──────────────────────────────────────
def _save_offline_cache(info: dict):
try:
info = dict(info)
info["_checked_at"] = time.time()
OFFLINE_CACHE_FILE.write_text(json.dumps(info), encoding="utf-8")
except Exception:
pass
def _load_offline_cache() -> dict | None:
"""Загрузить офлайн-кэш, если не истёк grace-период."""
if not OFFLINE_CACHE_FILE.exists():
return None
try:
info = json.loads(OFFLINE_CACHE_FILE.read_text(encoding="utf-8"))
checked = info.get("_checked_at", 0)
if time.time() - checked <= OFFLINE_GRACE_DAYS * 86400:
return info
except Exception:
pass
return None
# ── Multi-Mirror Failover HTTP-клиент ───────────────────────
def _try_endpoint(base: str, path: str, data: bytes | None,
headers: dict, timeout: int = 5) -> dict | None:
url = f"{base.rstrip('/')}{path}"
req = urllib.request.Request(url, data=data, headers=headers,
method="POST" if data else "GET")
with urllib.request.urlopen(req, timeout=timeout) as resp:
return json.loads(resp.read().decode("utf-8"))
def _request(path: str, payload: dict | None, key: str | None) -> dict:
"""HTTP-запрос с failover по API_ENDPOINTS (primary → зеркала → fallback)."""
data = json.dumps(payload).encode("utf-8") if payload is not None else None
headers = {"Content-Type": "application/json"} if data else {}
if key:
headers["X-License-Key"] = key
# Дедуплицируем endpoints, сохраняя порядок
seen = []
for ep in API_ENDPOINTS:
if ep not in seen:
seen.append(ep)
last_err = None
for base in seen:
try:
return _try_endpoint(base, path, data, headers)
except Exception as e:
last_err = e
continue
raise last_err or ConnectionError("Все зеркала недоступны")
def _post(path: str, payload: dict, key: str | None) -> dict:
result = _request(path, payload, key)
_save_offline_cache(result)
return result
def _get(path: str, key: str | None) -> dict:
result = _request(path, None, key)
_save_offline_cache(result)
return result
def get_license(key: str) -> dict:
"""Получить данные лицензии (с offline grace fallback)."""
try:
return _get("/cabinet/license", key)
except Exception:
# Offline Grace Mode: отдать кэш, если в пределах grace-периода
cached = _load_offline_cache()
if cached is not None:
return cached
raise
def activate(license_key: str, device_id: str, device_name: str | None = None) -> dict:
"""Активировать лицензию на этом устройстве."""
payload = {
"license_key": license_key,
"device_id": device_id,
"platform": "WINDOWS",
"device_name": device_name,
}
return _post("/client/activate", payload, license_key)
def heartbeat(key: str, device_id: str) -> dict:
"""Отправить heartbeat."""
payload = {"license_key": key, "device_id": device_id}
return _post("/client/heartbeat", payload, key)
def stack_key(current_key: str, new_key: str) -> dict:
"""Сложить ключи (продление)."""
payload = {"current_key": current_key, "new_key": new_key}
return _post("/cabinet/stack-key", payload, current_key)
def switch_key(current_key: str, new_key: str) -> dict:
"""Сменить тариф (upgrade/downgrade)."""
payload = {"current_key": current_key, "new_key": new_key}
return _post("/cabinet/switch-key", payload, current_key)
def get_device_id() -> str:
"""Получить стабильный ID устройства (MachineGUID на Windows)."""
import uuid
try:
import winreg
key = winreg.OpenKey(
winreg.HKEY_LOCAL_MACHINE,
r"SOFTWARE\Microsoft\Cryptography",
0,
winreg.KEY_READ | winreg.KEY_WOW64_64KEY,
)
val, _ = winreg.QueryValueEx(key, "MachineGuid")
winreg.CloseKey(key)
return str(val)
except Exception:
return str(uuid.getnode())
======================================================================МОДУЛЬ: Менеджер паролейФАЙЛ: windows-agent/core/vault.py======================================================================
# -*- coding: utf-8 -*-
"""Менеджер паролей (Vault): AES-256 локальное шифрование + синхронизация.
Zero-Knowledge: пароли шифруются локально мастер-паролем, сервер хранит только
зашифрованный слепок. Реализация через hashlib (PBKDF2) + XOR-stream (демо).
Для боевой версии заменить на AES-GCM (cryptography/pycryptodome).
"""
import base64
import hashlib
import json
import os
import secrets
import time
import urllib.request
from pathlib import Path
from core.config import API_BASE, VAULT_FILE
PBKDF2_ITERATIONS = 200_000
def _derive_key(master_password: str, salt: bytes) -> bytes:
"""PBKDF2-HMAC-SHA256 → 32-байтовый ключ."""
return hashlib.pbkdf2_hmac("sha256", master_password.encode(), salt, PBKDF2_ITERATIONS, dklen=32)
def _xor_stream(data: bytes, key: bytes) -> bytes:
"""XOR stream cipher (демо-реализация)."""
return bytes(b ^ key[i % len(key)] for i, b in enumerate(data))
def encrypt_blob(entries: list[dict], master_password: str) -> tuple[str, str]:
"""Зашифровать список записей. Возвращает (blob_b64, blob_hash)."""
salt = secrets.token_bytes(16)
key = _derive_key(master_password, salt)
plaintext = json.dumps(entries, ensure_ascii=False).encode("utf-8")
# Добавляем salt в начало
payload = salt + _xor_stream(plaintext, key)
blob_b64 = base64.b64encode(payload).decode("ascii")
blob_hash = hashlib.sha256(payload).hexdigest()
return blob_b64, blob_hash
def decrypt_blob(blob_b64: str, master_password: str) -> list[dict]:
"""Расшифровать слепок. Возвращает список записей."""
payload = base64.b64decode(blob_b64)
salt = payload[:16]
ciphertext = payload[16:]
key = _derive_key(master_password, salt)
plaintext = _xor_stream(ciphertext, key)
return json.loads(plaintext.decode("utf-8"))
def save_local(entries: list[dict]):
"""Сохранить записи локально (для автономного режима)."""
VAULT_FILE.write_text(json.dumps(entries, ensure_ascii=False, indent=2), encoding="utf-8")
def load_local() -> list[dict]:
"""Загрузить локальные записи."""
if VAULT_FILE.exists():
try:
return json.loads(VAULT_FILE.read_text(encoding="utf-8"))
except Exception:
return []
return []
def sync_to_server(license_key: str, master_password: str) -> dict:
"""Синхронизировать локальный vault на сервер (шифруя мастер-паролем)."""
entries = load_local()
blob_b64, blob_hash = encrypt_blob(entries, master_password)
url = f"{API_BASE}/vault/sync"
payload = json.dumps({
"encrypted_blob": blob_b64,
"blob_hash": blob_hash,
"entry_count": len(entries),
}).encode("utf-8")
headers = {"Content-Type": "application/json", "X-License-Key": license_key}
req = urllib.request.Request(url, data=payload, headers=headers, method="POST")
with urllib.request.urlopen(req, timeout=15) as resp:
return json.loads(resp.read().decode("utf-8"))
def sync_from_server(license_key: str, master_password: str) -> list[dict]:
"""Получить vault с сервера и расшифровать."""
url = f"{API_BASE}/vault/sync"
headers = {"X-License-Key": license_key}
req = urllib.request.Request(url, headers=headers, method="GET")
with urllib.request.urlopen(req, timeout=15) as resp:
data = json.loads(resp.read().decode("utf-8"))
blob = data.get("encrypted_blob")
if not blob:
return []
entries = decrypt_blob(blob, master_password)
save_local(entries)
return entries
======================================================================МОДУЛЬ: Кэш + белый списокФАЙЛ: windows-agent/core/cache.py======================================================================
# -*- coding: utf-8 -*-
"""SQLite Smart Cache — кэш проверенных файлов (size + mtime).
Если файл уже проверялся и его size/mtime не изменились — пропуск за 0 мс.
Это даёт ~0% нагрузки на CPU при фоновом мониторинге.
"""
import sqlite3
import threading
from pathlib import Path
from core.config import CACHE_DB
class ScanCache:
"""Потокобезопасный кэш результатов сканирования."""
def __init__(self, db_path: Path = CACHE_DB):
self.db_path = str(db_path)
self._local = threading.local()
self._init_db()
def _conn(self) -> sqlite3.Connection:
if not hasattr(self._local, "conn") or self._local.conn is None:
self._local.conn = sqlite3.connect(self.db_path)
self._local.conn.row_factory = sqlite3.Row
return self._local.conn
def _init_db(self):
conn = self._conn()
conn.execute(
"""CREATE TABLE IF NOT EXISTS scan_cache (
path TEXT PRIMARY KEY,
size INTEGER NOT NULL,
mtime REAL NOT NULL,
sha256 TEXT,
status TEXT NOT NULL,
threat_name TEXT,
scanned_at REAL NOT NULL
)"""
)
# Белый список исключений (Exclusions) — восстановленные/разрешённые файлы
conn.execute(
"""CREATE TABLE IF NOT EXISTS whitelist (
file_path TEXT PRIMARY KEY,
sha256 TEXT,
added_at REAL NOT NULL
)"""
)
conn.commit()
# ── Белый список (Exclusions) ──────────────────────────────
def is_whitelisted(self, path: Path, sha256: str | None = None) -> bool:
"""Есть ли путь или sha256 в белом списке. Если да — проверка пропускается (CLEAN)."""
conn = self._conn()
if conn.execute(
"SELECT 1 FROM whitelist WHERE file_path = ?", (str(path),)
).fetchone():
return True
if sha256:
row = conn.execute(
"SELECT 1 FROM whitelist WHERE sha256 = ?", (sha256,)
).fetchone()
if row:
return True
return False
def add_whitelist(self, path: Path, sha256: str | None = None):
"""Добавить файл в белый список."""
import time
conn = self._conn()
conn.execute(
"INSERT OR REPLACE INTO whitelist (file_path, sha256, added_at) VALUES (?, ?, ?)",
(str(path), sha256, time.time()),
)
conn.commit()
def remove_whitelist(self, path: str):
"""Удалить запись из белого списка по пути."""
conn = self._conn()
conn.execute("DELETE FROM whitelist WHERE file_path = ?", (path,))
conn.commit()
def list_whitelist(self) -> list[dict]:
"""Список исключений."""
conn = self._conn()
rows = conn.execute(
"SELECT file_path, sha256, added_at FROM whitelist ORDER BY added_at DESC"
).fetchall()
return [dict(r) for r in rows]
def get(self, path: Path) -> sqlite3.Row | None:
"""Получить запись кэша для файла (или None)."""
try:
st = path.stat()
except OSError:
return None
conn = self._conn()
row = conn.execute(
"SELECT * FROM scan_cache WHERE path = ?", (str(path),)
).fetchone()
if row is None:
return None
# Сверяем size + mtime — если изменился, кэш устарел
if row["size"] != st.st_size or abs(row["mtime"] - st.st_mtime) > 0.001:
return None
return row
def put(self, path: Path, sha256: str, status: str, threat_name: str | None = None):
"""Сохранить результат сканирования в кэш."""
import time
try:
st = path.stat()
except OSError:
return
conn = self._conn()
conn.execute(
"""INSERT OR REPLACE INTO scan_cache
(path, size, mtime, sha256, status, threat_name, scanned_at)
VALUES (?, ?, ?, ?, ?, ?, ?)""",
(str(path), st.st_size, st.st_mtime, sha256, status, threat_name, time.time()),
)
conn.commit()
def stats(self) -> dict:
"""Статистика кэша."""
conn = self._conn()
total = conn.execute("SELECT COUNT(*) FROM scan_cache").fetchone()[0]
malware = conn.execute(
"SELECT COUNT(*) FROM scan_cache WHERE status = 'MALWARE'"
).fetchone()[0]
return {"total": total, "malware": malware}
======================================================================МОДУЛЬ: Конфигурация агентаФАЙЛ: windows-agent/core/config.py======================================================================
# -*- coding: utf-8 -*-
"""Конфигурация Windows-агента AV Platform."""
import os
import sys
from pathlib import Path
APP_NAME = "AV Platform Agent"
APP_VERSION = "1.0.0"
# API backend (через SSH-туннель или прямой доступ)
API_BASE = os.environ.get("AV_API_BASE", "http://127.0.0.1:8450/api/v1")
# Админ-токен (для backup export/import и служебных операций)
ADMIN_TOKEN = os.environ.get("AV_ADMIN_TOKEN", "av-admin-token-7c4e9b2a8d1f6e3c5a0b")
# ── Multi-Mirror Geo-Failover ─────────────────────────────────
# Порядок: Primary → Mirror KZ → Mirror EU → локальный фолбэк.
# HTTP-клиент (license.py/backup) идёт по списку сверху вниз,
# пока не получит живой ответ (2xx). Мёртвый primary → живое зеркало.
API_ENDPOINTS = [
"http://127.0.0.1:8450", # PRIMARY (локальный/основной узел)
"https://api.cifra-secure.kz", # MIRROR KZ (standby)
"https://api.cifra-secure.eu", # MIRROR EU (standby)
"http://127.0.0.1:8450", # LOCAL FALLBACK
]
# Offline Grace Mode: сколько дней работать без связи с сервером
OFFLINE_GRACE_DAYS = 14
# Лицензионный ключ (хранится локально)
CONFIG_DIR = Path(os.environ.get("APPDATA", "~")) / "AVPlatformAgent"
CONFIG_DIR.mkdir(parents=True, exist_ok=True)
KEY_FILE = CONFIG_DIR / "license.key"
CACHE_DB = CONFIG_DIR / "scan_cache.sqlite"
QUARANTINE_DIR = CONFIG_DIR / "quarantine"
QUARANTINE_DIR.mkdir(parents=True, exist_ok=True)
VAULT_FILE = CONFIG_DIR / "vault.dat"
LOG_FILE = CONFIG_DIR / "agent.log"
OFFLINE_CACHE_FILE = CONFIG_DIR / "offline_license_cache.json"
# Папки для мониторинга (критические пути пользователя)
def critical_folders() -> list[Path]:
"""Критические папки для реального мониторинга."""
home = Path.home()
folders = [
home / "Downloads",
home / "AppData" / "Local" / "Temp",
home / "AppData" / "Roaming" / "Microsoft" / "Windows" / "Start Menu" / "Programs" / "Startup",
home / "AppData" / "Local",
]
return [f for f in folders if f.exists()]
# Порог: не сканировать файлы больше N МБ (эвристика)
MAX_SCAN_SIZE_MB = 100
# Интервал фонового heartbeat (сек)
HEARTBEAT_INTERVAL = 300
======================================================================МОДУЛЬ: GUI агентаФАЙЛ: windows-agent/agent.py======================================================================
# -*- coding: utf-8 -*-
"""AV Platform Agent — сверхлёгкий Windows-агент (User-Mode, «Установил и забыл»).
GUI в стиле ESET HOME: 920x600, тёмная/светлая тема, RU/EN, 4 вкладки.
"""
import os
import sys
import threading
import tkinter as tk
from pathlib import Path
from tkinter import ttk, messagebox, simpledialog
# Ядро агента
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from core import config
from core.cache import ScanCache
from core.scanner import scan_file
from core.monitor import FileMonitor
from core import quarantine
from core import license as lic
from core import vault
# ── Цвета (ESET HOME) ────────────────────────────────────────
DARK = {"bg": "#0B1120", "card": "#1E293B", "fg": "#E2E8F0", "muted": "#94A3B8",
"accent": "#10B981", "border": "#334155"}
LIGHT = {"bg": "#F8FAFC", "card": "#FFFFFF", "fg": "#1E293B", "muted": "#64748B",
"accent": "#10B981", "border": "#E2E8F0"}
I18N = {
"ru": {
"app": "CIFRA SECURE • Global Edition",
"tab_overview": "Обзор",
"tab_scan": "Проверка и Карантин",
"tab_vault": "Менеджер паролей",
"tab_license": "Лицензия и Настройки",
"status_active": "Защита активна",
"status_inactive": "Защита неактивна",
"shield_ok": "🛡️ Защищено",
"shield_off": "⚠️ Не защищено",
"ram": "Потребление ОЗУ",
"quick_scan": "Быстрая проверка",
"scanning": "Проверка…",
"scanned": "Проверено файлов",
"threats": "Угроз найдено",
"quarantine": "Карантин",
"restore": "Восстановить",
"delete": "Удалить",
"no_threats": "Угроз не обнаружено",
"vault_title": "Менеджер паролей (AES-256)",
"vault_master": "Мастер-пароль",
"vault_add": "+ Добавить запись",
"vault_sync": "Синхронизировать с облаком",
"vault_site": "Сайт",
"vault_login": "Логин",
"vault_pass": "Пароль",
"license_key": "Лицензионный ключ",
"license_activate": "Активировать",
"license_stack": "Суммировать ключ",
"license_switch": "Сменить тариф",
"theme": "Тема",
"language": "Язык",
"who_calls": "Защита звонков (Who Calls) & Антиспам",
"privacy_shield": "Монитор приватности (Privacy Shield)",
"privacy_lines": "Анонимная активация (Без Email, телефона и анкет)|Слежка и телеметрия: 0% (Отключены на уровне ядра)|Личные файлы не покидают устройство (Проверка только по SHA-256)",
"zero_telemetry": "🕵️♂️ ZERO-TELEMETRY (Без слежки и сбора данных)",
"restore_whitelist": "✅ Восстановить + В белый список",
"exclusions_title": "Исключения (Белый список)",
"excl_path": "Путь",
"excl_remove": "Удалить из списка",
"restored_whitelisted_msg": "Файл восстановлен и добавлен в белый список. Проверка пропускается.",
"pause_protection": "⏸️ Пауза защиты",
"pause_15m": "Пауза: 15 минут",
"pause_1h": "Пауза: 1 час",
"pause_reboot": "Пауза: До перезагрузки",
"resume_protection": "▶️ Возобновить защиту",
"shield_paused": "⏸️ Защита на паузе",
"paused_until": "Пауза до",
"device": "Устройство",
"days_left": "Осталось дней",
"tariff": "Тариф",
},
"en": {
"app": "CIFRA SECURE • Global Edition",
"tab_overview": "Overview",
"tab_scan": "Scan & Quarantine",
"tab_vault": "Password Manager",
"tab_license": "License & Settings",
"status_active": "Protection Active",
"status_inactive": "Protection Inactive",
"shield_ok": "🛡️ Protected",
"shield_off": "⚠️ Not Protected",
"ram": "RAM Usage",
"quick_scan": "Quick Scan",
"scanning": "Scanning…",
"scanned": "Files Scanned",
"threats": "Threats Found",
"quarantine": "Quarantine",
"restore": "Restore",
"delete": "Delete",
"no_threats": "No threats detected",
"vault_title": "Password Manager (AES-256)",
"vault_master": "Master Password",
"vault_add": "+ Add Entry",
"vault_sync": "Sync with Cloud",
"vault_site": "Site",
"vault_login": "Login",
"vault_pass": "Password",
"license_key": "License Key",
"license_activate": "Activate",
"license_stack": "Stack Key",
"license_switch": "Switch Plan",
"theme": "Theme",
"language": "Language",
"who_calls": "Secure VPN & Caller ID",
"privacy_shield": "Privacy Shield",
"privacy_lines": "Anonymous activation (No email, phone, or forms)|Tracking & telemetry: 0% (Disabled at core level)|Personal files never leave device (SHA-256 check only)",
"zero_telemetry": "🕵️♂️ ZERO-TELEMETRY (100% Private)",
"restore_whitelist": "✅ Restore & Whitelist",
"exclusions_title": "Exclusions (Whitelist)",
"excl_path": "Path",
"excl_remove": "Remove from list",
"restored_whitelisted_msg": "File restored and added to whitelist. Scan will be skipped.",
"pause_protection": "⏸️ Pause Protection",
"pause_15m": "Pause: 15 minutes",
"pause_1h": "Pause: 1 hour",
"pause_reboot": "Pause: Until Reboot",
"resume_protection": "▶️ Resume Protection",
"shield_paused": "⏸️ Protection Paused",
"paused_until": "Paused until",
"device": "Device",
"days_left": "Days Left",
"tariff": "Plan",
},
}
class AgentApp(tk.Tk):
"""Главное окно агента."""
def __init__(self):
super().__init__()
self.lang = "ru"
self.theme_name = "dark"
self.colors = DARK
self.title(I18N[self.lang]["app"])
self.geometry("920x600")
self.minsize(920, 600)
self.configure(bg=self.colors["bg"])
self.cache = ScanCache()
self.device_id = lic.get_device_id()
self.license_key = lic.load_key()
self.monitor = None
self._build_ui()
self._apply_theme()
def t(self, key):
return I18N[self.lang].get(key, key)
# ── Построение UI ────────────────────────────────────────
def _build_ui(self):
# Левая панель (вкладки)
self.sidebar = tk.Frame(self, width=200, bg=self.colors["card"])
self.sidebar.pack(side="left", fill="y")
self.sidebar.pack_propagate(False)
self.logo = tk.Label(self.sidebar, text="A", font=("Segoe UI", 24, "bold"),
bg=self.colors["accent"], fg="white", width=2)
self.logo.pack(pady=(20, 5))
self.tab_buttons = {}
for i, key in enumerate(["tab_overview", "tab_scan", "tab_vault", "tab_license"]):
btn = tk.Button(self.sidebar, text=self.t(key), anchor="w",
font=("Segoe UI", 11), relief="flat", bd=0,
padx=20, pady=12, cursor="hand2",
command=lambda k=key: self.show_tab(k))
btn.pack(fill="x")
self.tab_buttons[key] = btn
# Переключатель темы/языка внизу сайдбара
ctrl = tk.Frame(self.sidebar, bg=self.colors["card"])
ctrl.pack(side="bottom", fill="x", padx=10, pady=10)
self.theme_btn = tk.Button(ctrl, text="◐", font=("Segoe UI", 14),
relief="flat", bd=0, cursor="hand2",
command=self.toggle_theme)
self.theme_btn.pack(side="left", padx=(0, 5))
self.lang_btn = tk.Button(ctrl, text="RU/EN", font=("Segoe UI", 9),
relief="flat", bd=0, cursor="hand2",
command=self.toggle_lang)
self.lang_btn.pack(side="left")
# Правая область (контент вкладок)
self.content = tk.Frame(self, bg=self.colors["bg"])
self.content.pack(side="right", fill="both", expand=True)
self.frames = {}
self._build_overview()
self._build_scan()
self._build_vault()
self._build_license()
self.show_tab("tab_overview")
def _card(self, parent):
return tk.Frame(parent, bg=self.colors["card"], padx=20, pady=16,
highlightbackground=self.colors["border"], highlightthickness=1)
def _build_overview(self):
f = tk.Frame(self.content, bg=self.colors["bg"], padx=24, pady=24)
self.frames["tab_overview"] = f
# Щит статуса
self.shield_label = tk.Label(f, text=self.t("shield_off"),
font=("Segoe UI", 28, "bold"),
bg=self.colors["bg"], fg=self.colors["fg"])
self.shield_label.pack(pady=(10, 4))
self.status_label = tk.Label(f, text=self.t("status_inactive"),
font=("Segoe UI", 14),
bg=self.colors["bg"], fg=self.colors["muted"])
self.status_label.pack()
# Карточки статистики
stats = tk.Frame(f, bg=self.colors["bg"])
stats.pack(fill="x", pady=20)
self.ram_var = tk.StringVar(value="— МБ")
self.scanned_var = tk.StringVar(value="0")
self.threats_var = tk.StringVar(value="0")
self._stat_card(stats, self.t("ram"), self.ram_var, 0)
self._stat_card(stats, self.t("scanned"), self.scanned_var, 1)
self._stat_card(stats, self.t("threats"), self.threats_var, 2)
# Кнопка быстрой проверки
self.scan_btn = tk.Button(f, text=self.t("quick_scan"),
font=("Segoe UI", 12, "bold"),
bg=self.colors["accent"], fg="white",
relief="flat", padx=24, pady=10, cursor="hand2",
command=self.quick_scan)
self.scan_btn.pack(pady=10)
# Пауза защиты (ESET-style dropdown)
pause_row = tk.Frame(f, bg=self.colors["bg"])
pause_row.pack(pady=(2, 0))
self.pause_btn = tk.Button(pause_row, text=self.t("pause_protection"),
font=("Segoe UI", 11),
bg=self.colors["card"], fg=self.colors["fg"],
relief="flat", padx=16, pady=6, cursor="hand2",
command=self._show_pause_menu)
self.pause_btn.pack(side="left", padx=4)
self.pause_countdown = tk.Label(pause_row, text="", font=("Segoe UI", 10),
bg=self.colors["bg"], fg="#F59E0B")
self.pause_countdown.pack(side="left", padx=4)
self._pause_menu = None
# Who Calls / VPN (условная локализация)
self.feature_label = tk.Label(f, text="📞 " + self.t("who_calls"),
font=("Segoe UI", 11),
bg=self.colors["bg"], fg=self.colors["accent"])
self.feature_label.pack(pady=(10, 0))
# Privacy Shield (Zero-Telemetry)
shield_card = tk.Frame(f, bg=self.colors["card"], padx=16, pady=12,
highlightbackground=self.colors["border"], highlightthickness=1)
shield_card.pack(fill="x", pady=(16, 0))
tk.Label(shield_card, text="🕵️♂️ " + self.t("privacy_shield"),
font=("Segoe UI", 11, "bold"),
bg=self.colors["card"], fg=self.colors["fg"]).pack(anchor="w", pady=(0, 6))
for line in self.t("privacy_lines").split("|"):
tk.Label(shield_card, text="✔️ " + line.strip(),
font=("Segoe UI", 9),
bg=self.colors["card"], fg=self.colors["accent"]).pack(anchor="w", pady=1)
# Zero-Telemetry badge (статус-бар внизу)
self.badge_label = tk.Label(f, text=self.t("zero_telemetry"),
font=("Segoe UI", 9),
bg=self.colors["bg"], fg=self.colors["muted"])
self.badge_label.pack(side="bottom", pady=(10, 0))
def _stat_card(self, parent, title, var, col):
card = tk.Frame(parent, bg=self.colors["card"], padx=16, pady=12,
highlightbackground=self.colors["border"], highlightthickness=1)
card.grid(row=0, column=col, padx=6, sticky="nsew")
parent.columnconfigure(col, weight=1)
tk.Label(card, text=title, font=("Segoe UI", 9),
bg=self.colors["card"], fg=self.colors["muted"]).pack(anchor="w")
tk.Label(card, textvariable=var, font=("Segoe UI", 18, "bold"),
bg=self.colors["card"], fg=self.colors["fg"]).pack(anchor="w")
def _build_scan(self):
f = tk.Frame(self.content, bg=self.colors["bg"], padx=24, pady=24)
self.frames["tab_scan"] = f
tk.Label(f, text=self.t("quarantine"), font=("Segoe UI", 16, "bold"),
bg=self.colors["bg"], fg=self.colors["fg"]).pack(anchor="w", pady=(0, 10))
cols = ("name", "path", "action")
self.q_tree = ttk.Treeview(f, columns=cols, show="headings", height=14)
self.q_tree.heading("name", text="Имя")
self.q_tree.heading("path", text="Путь")
self.q_tree.heading("action", text="")
self.q_tree.column("name", width=180)
self.q_tree.column("path", width=420)
self.q_tree.column("action", width=120)
self.q_tree.pack(fill="both", expand=True)
btns = tk.Frame(f, bg=self.colors["bg"])
btns.pack(fill="x", pady=8)
tk.Button(btns, text=self.t("restore"), command=self.restore_selected,
relief="flat", padx=12, pady=6, cursor="hand2").pack(side="left", padx=4)
tk.Button(btns, text=self.t("restore_whitelist"), command=self.restore_and_whitelist,
bg="#10B981", fg="white", relief="flat", padx=12, pady=6,
cursor="hand2").pack(side="left", padx=4)
tk.Button(btns, text=self.t("delete"), command=self.delete_selected,
relief="flat", padx=12, pady=6, cursor="hand2").pack(side="left", padx=4)
tk.Button(btns, text="Обновить", command=self.refresh_quarantine,
relief="flat", padx=12, pady=6, cursor="hand2").pack(side="right", padx=4)
self.refresh_quarantine()
# ── Список исключений (Exclusions / белый список) ────────
tk.Label(f, text=self.t("exclusions_title"), font=("Segoe UI", 13, "bold"),
bg=self.colors["bg"], fg=self.colors["fg"]).pack(anchor="w", pady=(14, 6))
ecols = ("epath", "esha")
self.ex_tree = ttk.Treeview(f, columns=ecols, show="headings", height=6)
self.ex_tree.heading("epath", text=self.t("excl_path"))
self.ex_tree.heading("esha", text="SHA-256")
self.ex_tree.column("epath", width=420)
self.ex_tree.column("esha", width=220)
self.ex_tree.pack(fill="both", expand=False)
exbtns = tk.Frame(f, bg=self.colors["bg"])
exbtns.pack(fill="x", pady=4)
tk.Button(exbtns, text=self.t("excl_remove"), command=self.remove_exclusion,
relief="flat", padx=12, pady=6, cursor="hand2").pack(side="left", padx=4)
tk.Button(exbtns, text="Обновить", command=self.refresh_exclusions,
relief="flat", padx=12, pady=6, cursor="hand2").pack(side="left", padx=4)
self.refresh_exclusions()
def _build_vault(self):
f = tk.Frame(self.content, bg=self.colors["bg"], padx=24, pady=24)
self.frames["tab_vault"] = f
tk.Label(f, text=self.t("vault_title"), font=("Segoe UI", 16, "bold"),
bg=self.colors["bg"], fg=self.colors["fg"]).pack(anchor="w", pady=(0, 10))
# Мастер-пароль
mrow = tk.Frame(f, bg=self.colors["bg"])
mrow.pack(fill="x", pady=4)
tk.Label(mrow, text=self.t("vault_master"), bg=self.colors["bg"],
fg=self.colors["muted"]).pack(side="left")
self.master_entry = tk.Entry(mrow, show="*", width=24)
self.master_entry.pack(side="right")
# Список записей
self.vault_list = tk.Listbox(f, height=12, font=("Consolas", 10))
self.vault_list.pack(fill="both", expand=True, pady=8)
# Кнопки
vb = tk.Frame(f, bg=self.colors["bg"])
vb.pack(fill="x")
tk.Button(vb, text=self.t("vault_add"), command=self.vault_add,
relief="flat", padx=12, pady=6, cursor="hand2").pack(side="left", padx=4)
tk.Button(vb, text=self.t("vault_sync"), command=self.vault_sync,
relief="flat", padx=12, pady=6, cursor="hand2").pack(side="left", padx=4)
self._refresh_vault_list()
def _build_license(self):
f = tk.Frame(self.content, bg=self.colors["bg"], padx=24, pady=24)
self.frames["tab_license"] = f
tk.Label(f, text=self.t("license_key"), font=("Segoe UI", 16, "bold"),
bg=self.colors["bg"], fg=self.colors["fg"]).pack(anchor="w", pady=(0, 10))
self.key_entry = tk.Entry(f, font=("Consolas", 12), justify="center")
self.key_entry.pack(fill="x", pady=6)
if self.license_key:
self.key_entry.insert(0, self.license_key)
tk.Button(f, text=self.t("license_activate"), command=self.activate_key,
bg=self.colors["accent"], fg="white", relief="flat",
padx=16, pady=8, cursor="hand2").pack(pady=6)
# Инфо о лицензии
self.lic_info = tk.Label(f, text="", font=("Segoe UI", 11),
bg=self.colors["bg"], fg=self.colors["muted"],
justify="left")
self.lic_info.pack(anchor="w", pady=8)
# Суммировать / Сменить тариф
row = tk.Frame(f, bg=self.colors["bg"])
row.pack(fill="x", pady=6)
tk.Button(row, text=self.t("license_stack"), command=self.stack_key,
relief="flat", padx=12, pady=6, cursor="hand2").pack(side="left", padx=4)
tk.Button(row, text=self.t("license_switch"), command=self.switch_key,
relief="flat", padx=12, pady=6, cursor="hand2").pack(side="left", padx=4)
self._refresh_license_info()
# ── Логика ───────────────────────────────────────────────
def show_tab(self, key):
for k, frame in self.frames.items():
frame.pack_forget()
self.tab_buttons[k].configure(bg=self.colors["card"], fg=self.colors["fg"])
self.frames[key].pack(fill="both", expand=True)
self.tab_buttons[key].configure(bg=self.colors["accent"], fg="white")
def toggle_theme(self):
self.theme_name = "light" if self.theme_name == "dark" else "dark"
self.colors = LIGHT if self.theme_name == "light" else DARK
self._apply_theme()
def toggle_lang(self):
self.lang = "en" if self.lang == "ru" else "ru"
self._refresh_texts()
def _apply_theme(self):
self.configure(bg=self.colors["bg"])
self.sidebar.configure(bg=self.colors["card"])
self.content.configure(bg=self.colors["bg"])
self.logo.configure(bg=self.colors["accent"])
for btn in self.tab_buttons.values():
btn.configure(bg=self.colors["card"], fg=self.colors["fg"],
activebackground=self.colors["accent"], activeforeground="white")
# Перерисовать содержимое
self._refresh_texts()
def _refresh_texts(self):
self.title(self.t("app"))
for k, btn in self.tab_buttons.items():
btn.configure(text=self.t(k))
# Обновляем статус
self._update_status()
def _update_status(self):
if self.license_key:
self.shield_label.configure(text=self.t("shield_ok"), fg=self.colors["accent"])
self.status_label.configure(text=self.t("status_active"))
else:
self.shield_label.configure(text=self.t("shield_off"))
self.status_label.configure(text=self.t("status_inactive"))
def quick_scan(self):
"""Быстрая проверка критических папок (в фоне)."""
self.scan_btn.configure(state="disabled", text=self.t("scanning"))
self.scanned_var.set("0")
self.threats_var.set("0")
def worker():
scanned = 0
threats = 0
for folder in config.critical_folders():
for path in folder.rglob("*"):
if path.is_file():
result = scan_file(path, self.license_key)
scanned += 1
if result["status"] == "MALWARE":
threats += 1
quarantine.quarantine_file(path)
self.after(0, lambda s=scanned, th=threats: (
self.scanned_var.set(str(s)),
self.threats_var.set(str(th)),
))
self.after(0, lambda: (
self.scan_btn.configure(state="normal", text=self.t("quick_scan")),
self.refresh_quarantine(),
messagebox.showinfo(self.t("quick_scan"),
f"{self.t('scanned')}: {scanned}\n{self.t('threats')}: {threats}"),
))
threading.Thread(target=worker, daemon=True).start()
def refresh_quarantine(self):
for i in self.q_tree.get_children():
self.q_tree.delete(i)
for item in quarantine.list_quarantine():
self.q_tree.insert("", "end", iid=item["quarantine_name"],
values=(item["original_name"], item["original_path"], ""))
def restore_selected(self):
sel = self.q_tree.selection()
if sel:
quarantine.restore_file(sel[0])
self.refresh_quarantine()
def delete_selected(self):
sel = self.q_tree.selection()
if sel and messagebox.askyesno(self.t("delete"), sel[0]):
quarantine.delete_from_quarantine(sel[0])
self.refresh_quarantine()
# ── Белый список (Exclusions) ──────────────────────────────
def restore_and_whitelist(self):
"""Восстановить из карантина И сразу добавить в белый список (1 клик)."""
sel = self.q_tree.selection()
if not sel:
return
res = quarantine.restore_file(sel[0])
if res.get("success"):
restored = Path(res["restored_to"])
# sha256 уже известен? берём из кэша, иначе compute
sha = None
cached = self.cache.get(restored)
if cached is not None:
sha = cached["sha256"]
self.cache.add_whitelist(restored, sha)
# Сбрасываем статус в кэше, чтобы файл считался CLEAN
messagebox.showinfo(self.t("restore_whitelist"),
self.t("restored_whitelisted_msg"))
self.refresh_quarantine()
self.refresh_exclusions()
def refresh_exclusions(self):
for i in self.ex_tree.get_children():
self.ex_tree.delete(i)
for item in self.cache.list_whitelist():
sha = (item.get("sha256") or "")[:16]
self.ex_tree.insert("", "end", iid=item["file_path"],
values=(item["file_path"], sha))
def remove_exclusion(self):
sel = self.ex_tree.selection()
if sel:
self.cache.remove_whitelist(sel[0])
self.refresh_exclusions()
# ── Пауза защиты (ESET-style) ──────────────────────────────
def _show_pause_menu(self):
"""Выпадающее меню паузы защиты."""
menu = tk.Menu(self, tearoff=0, bg=self.colors["card"], fg=self.colors["fg"])
if self.monitor and self.monitor.is_paused():
menu.add_command(label=self.t("resume_protection"),
command=self.resume_protection)
else:
menu.add_command(label=self.t("pause_15m"),
command=lambda: self.pause_protection(15 * 60))
menu.add_command(label=self.t("pause_1h"),
command=lambda: self.pause_protection(60 * 60))
menu.add_command(label=self.t("pause_reboot"),
command=lambda: self.pause_protection(None))
# Позиционировать под кнопкой
x = self.pause_btn.winfo_rootx()
y = self.pause_btn.winfo_rooty() + self.pause_btn.winfo_height()
menu.tk_popup(x, y)
def pause_protection(self, seconds: float | None):
"""Поставить защиту на паузу (seconds=None → до перезагрузки)."""
if self.monitor:
self.monitor.pause(seconds)
self._update_shield()
self._tick_pause()
def resume_protection(self):
"""Возобновить защиту."""
if self.monitor:
self.monitor.resume()
self.pause_countdown.config(text="")
self._update_shield()
def _update_shield(self):
"""Обновить щит: зелёный (активна) / амбер #F59E0B (пауза) / серый (выкл)."""
if self.monitor and self.monitor.is_paused():
self.shield_label.config(text=self.t("shield_paused"), fg="#F59E0B")
self.status_label.config(text=self.t("pause_protection"), fg="#F59E0B")
elif self.monitor and self.monitor._thread and self.monitor._thread.is_alive():
self.shield_label.config(text=self.t("shield_ok"), fg="#10B981")
self.status_label.config(text=self.t("status_active"), fg=self.colors["muted"])
else:
self.shield_label.config(text=self.t("shield_off"), fg=self.colors["fg"])
self.status_label.config(text=self.t("status_inactive"), fg=self.colors["muted"])
def _tick_pause(self):
"""Таймер обратного отсчёта паузы (обновление раз в секунду)."""
if self.monitor and self.monitor.is_paused():
rem = self.monitor.pause_remaining()
if rem == float("inf"):
self.pause_countdown.config(text="⏸ " + self.t("pause_reboot"))
else:
m, s = divmod(int(rem), 60)
h, m = divmod(m, 60)
self.pause_countdown.config(text=f"⏸ {h:02d}:{m:02d}:{s:02d}")
self.after(1000, self._tick_pause)
else:
self.pause_countdown.config(text="")
self._update_shield()
def _refresh_vault_list(self):
self.vault_list.delete(0, "end")
for e in vault.load_local():
self.vault_list.insert("end", f"{e.get('site','')} · {e.get('login','')}")
def vault_add(self):
site = simpledialog.askstring(self.t("vault_site"), self.t("vault_site"))
if not site:
return
login = simpledialog.askstring(self.t("vault_login"), self.t("vault_login"))
pwd = simpledialog.askstring(self.t("vault_pass"), self.t("vault_pass"), show="*")
entries = vault.load_local()
entries.append({"site": site, "login": login or "", "password": pwd or ""})
vault.save_local(entries)
self._refresh_vault_list()
def vault_sync(self):
mp = self.master_entry.get()
if not mp:
messagebox.showwarning(self.t("vault_master"), self.t("vault_master"))
return
if not self.license_key:
messagebox.showwarning(self.t("license_key"), self.t("license_key"))
return
try:
vault.sync_to_server(self.license_key, mp)
messagebox.showinfo(self.t("vault_sync"), "OK")
except Exception as e:
messagebox.showerror(self.t("vault_sync"), str(e))
def activate_key(self):
key = self.key_entry.get().strip()
if len(key) < 19:
messagebox.showwarning(self.t("license_key"), self.t("license_key"))
return
try:
lic.activate(key, self.device_id, "Windows PC")
self.license_key = key
lic.save_key(key)
self._save_key_backup(key)
self._update_status()
self._refresh_license_info()
messagebox.showinfo(self.t("license_activate"), self.t("status_active"))
except Exception as e:
messagebox.showerror(self.t("license_activate"), str(e))
def _save_key_backup(self, key: str):
"""Авто-сохранить памятку CIFRA_SECURE_KEY_BACKUP.txt (+ на D:\\ если есть)."""
content = (
"=== CIFRA SECURE | СЕРТИФИКАТ ЛИЦЕНЗИИ И ВОССТАНОВЛЕНИЯ ===\n\n"
f"Ваш ключ активации: {key}\n"
"Устройство (HWID): " + self.device_id + "\n\n"
"Способы восстановления (без Email):\n"
" 1) Авто-по железу (HWID) — запустите агент без ключа на этом же ПК.\n"
" 2) По числу восстановления XXX-XXX-XX.\n"
" 3) По связке: слово + 4 цифры.\n\n"
"⚠️ Сохраните этот файл! Мы не храним ваши личные данные (Zero-Telemetry).\n"
"Активируйте ключ в день покупки.\nCIFRA SECURE • Global Edition\n"
)
targets = [Path.home() / "CIFRA_SECURE_KEY_BACKUP.txt"]
d_drive = Path("D:/")
if d_drive.exists():
targets.append(d_drive / "CIFRA_SECURE_KEY_BACKUP.txt")
for tpath in targets:
try:
tpath.write_text(content, encoding="utf-8")
except Exception:
pass
def stack_key(self):
new_key = simpledialog.askstring(self.t("license_stack"), "XXXX-XXXX-XXXX-XXXX")
if not new_key or not self.license_key:
return
try:
lic.stack_key(self.license_key, new_key.strip())
self._refresh_license_info()
messagebox.showinfo(self.t("license_stack"), "OK")
except Exception as e:
messagebox.showerror(self.t("license_stack"), str(e))
def switch_key(self):
new_key = simpledialog.askstring(self.t("license_switch"), "XXXX-XXXX-XXXX-XXXX")
if not new_key or not self.license_key:
return
try:
lic.switch_key(self.license_key, new_key.strip())
self._refresh_license_info()
messagebox.showinfo(self.t("license_switch"), "OK")
except Exception as e:
messagebox.showerror(self.t("license_switch"), str(e))
def _refresh_license_info(self):
if not self.license_key:
self.lic_info.configure(text="")
return
try:
info = lic.get_license(self.license_key)
edition = info.get("edition", "?")
days = "—"
self.lic_info.configure(
text=f"{self.t('tariff')}: {edition} · {self.t('device')}: {self.device_id[:8]}…"
)
except Exception as e:
self.lic_info.configure(text=str(e))
def main():
app = AgentApp()
# Запуск фонового мониторинга
app.monitor = FileMonitor(app.cache, app.license_key)
app.monitor.start()
# Обновление счётчиков ОЗУ
def update_ram():
try:
import psutil
mb = psutil.Process(os.getpid()).memory_info().rss / (1024 * 1024)
app.ram_var.set(f"{mb:.1f} МБ")
except Exception:
app.ram_var.set("— МБ")
app.after(3000, update_ram)
app.after(1000, update_ram)
# ── Системный трей (pystray + Pillow, с fallback) ──────────
def setup_tray():
try:
import threading
import pystray
from PIL import Image, ImageDraw
# Рисуем простой shield-икон 64x64
img = Image.new("RGB", (64, 64), "#0B1120")
d = ImageDraw.Draw(img)
d.polygon([(32, 8), (54, 18), (54, 34), (32, 56), (10, 34), (10, 18)],
fill="#10B981")
def open_window(icon, item):
app.after(0, app.deiconify)
def pause(sec):
def _h(icon=None, item=None):
app.after(0, lambda: app.pause_protection(sec))
return _h
def resume(icon=None, item=None):
app.after(0, app.resume_protection)
menu = pystray.Menu(
pystray.MenuItem(app.t("tab_overview"), open_window, default=True),
pystray.Menu.SEPARATOR,
pystray.MenuItem(app.t("pause_15m"), pause(15 * 60)),
pystray.MenuItem(app.t("pause_1h"), pause(60 * 60)),
pystray.MenuItem(app.t("pause_reboot"), pause(None)),
pystray.MenuItem(app.t("resume_protection"), resume),
)
icon = pystray.Icon("AVPlatform", img, app.t("app"), menu)
# Минимизировать в трей при закрытии окна
app.protocol("WM_DELETE_WINDOW", lambda: (app.withdraw(), None))
threading.Thread(target=icon.run, daemon=True).start()
except Exception:
# Fallback: трей недоступен (headless/нет pystray) — работаем без него
pass
setup_tray()
app.mainloop()
if __name__ == "__main__":
main()