Тесты и CI были добавлены

Каркас pytest с изоляцией путей в tmp: 40 тестов на валидаторы имён,
защиту от path traversal, парсеры virsh-вывода, cloud-init генерацию,
ротацию бэкапов, CSRF, rate-limit и рендеринг ключевых страниц.
GitHub Actions прогоняет compileall, bash -n всех скриптов и pytest.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01PKUoZwHMCHnytfswFVRUHw

.github/workflows/ci.yml

@@ -0,0 +1,31 @@
+name: CI
+
+on:
+  push:
+  pull_request:
+
+jobs:
+  checks:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+
+      - uses: actions/setup-python@v5
+        with:
+          python-version: "3.12"
+
+      - name: Проверка синтаксиса Python
+        run: python -m compileall -q web scripts
+
+      - name: Проверка синтаксиса bash-скриптов
+        run: |
+          bash -n install.sh bootstrap.sh install_virtuality_node.sh setup_github_sync.sh
+          for script in scripts/*.sh; do bash -n "$script"; done
+
+      - name: Установка зависимостей
+        run: |
+          python -m pip install --upgrade pip
+          pip install -r web/requirements.txt pytest httpx
+
+      - name: Тесты
+        run: pytest -q

tests/conftest.py

@@ -0,0 +1,38 @@
+import sys
+from pathlib import Path
+
+import pytest
+
+WEB_DIR = Path(__file__).resolve().parent.parent / "web"
+sys.path.insert(0, str(WEB_DIR))
+
+
+@pytest.fixture(autouse=True)
+def isolate_dirs(tmp_path, monkeypatch):
+    """Уводим все записи на диск в tmp, чтобы тесты работали без root и не трогали систему."""
+    import app
+    import backup_core
+    import cloud_images
+
+    monkeypatch.setattr(app, "OPERATIONS_DIR", tmp_path / "ops")
+    monkeypatch.setattr(app, "ISO_DIR", tmp_path / "iso")
+    monkeypatch.setattr(app, "IMAGES_DIR", tmp_path / "images")
+    monkeypatch.setattr(app, "DISK_IMAGES_DIR", tmp_path / "disk-images")
+    monkeypatch.setattr(app, "VM_TEMPLATES_FILE", tmp_path / "config" / "vm_templates.json")
+    monkeypatch.setattr(backup_core, "BACKUPS_DIR", tmp_path / "backups")
+    monkeypatch.setattr(backup_core, "SCHEDULE_FILE", tmp_path / "config" / "backup_schedule.json")
+    monkeypatch.setattr(backup_core, "LOG_FILE", tmp_path / "log" / "backup.log")
+    monkeypatch.setattr(cloud_images, "CLOUD_IMAGES_DIR", tmp_path / "cloud-images")
+    yield
+
+
+@pytest.fixture()
+def auth_client():
+    """TestClient с валидной сессией и CSRF-токеном."""
+    from fastapi.testclient import TestClient
+    import app
+
+    client = TestClient(app.app)
+    token = app.serializer.dumps({"user": app.AUTH_USER, "csrf": "test-csrf-token"})
+    client.cookies.set("virtuality_session", token)
+    return client

tests/test_app_security.py

@@ -0,0 +1,82 @@
+import app
+
+
+def test_login_page_renders(auth_client):
+    response = auth_client.get("/login", follow_redirects=False)
+    # авторизованный пользователь уводится на дашборд
+    assert response.status_code == 303
+
+
+def test_unauthenticated_post_redirects_to_login():
+    from fastapi.testclient import TestClient
+
+    client = TestClient(app.app)
+    response = client.post("/host/refresh", follow_redirects=False)
+    assert response.status_code == 303
+    assert response.headers["location"] == "/login"
+
+
+def test_post_without_csrf_rejected(auth_client):
+    response = auth_client.post("/iso/refresh", follow_redirects=False)
+    assert response.status_code == 403
+
+
+def test_post_with_wrong_csrf_rejected(auth_client):
+    response = auth_client.post("/iso/refresh", data={"csrf_token": "wrong"}, follow_redirects=False)
+    assert response.status_code == 403
+
+
+def test_post_with_valid_csrf_accepted(auth_client):
+    response = auth_client.post("/iso/refresh", data={"csrf_token": "test-csrf-token"}, follow_redirects=False)
+    assert response.status_code == 303
+
+
+def test_dashboard_renders_with_csrf_inputs(auth_client):
+    response = auth_client.get("/")
+    assert response.status_code == 200
+    assert 'name="csrf_token"' in response.text
+    assert "cpu-spark" in response.text
+
+
+def test_backups_page_renders(auth_client):
+    response = auth_client.get("/backups")
+    assert response.status_code == 200
+
+
+def test_cloud_images_page_renders(auth_client):
+    response = auth_client.get("/cloud-images")
+    assert response.status_code == 200
+
+
+def test_login_rate_limit():
+    ip = "203.0.113.7"
+    app.reset_login_failures(ip)
+    assert app.login_block_remaining(ip) == 0
+    for _ in range(app.LOGIN_MAX_FAILURES):
+        app.register_login_failure(ip)
+    assert app.login_block_remaining(ip) > 0
+    app.reset_login_failures(ip)
+    assert app.login_block_remaining(ip) == 0
+
+
+def test_session_token_roundtrip():
+    token = app.serializer.dumps({"user": app.AUTH_USER, "csrf": "abc"})
+    assert app.user_from_session_token(token) == app.AUTH_USER
+    assert app.user_from_session_token("garbage") is None
+    assert app.user_from_session_token(None) is None
+
+
+def test_template_vm_cannot_start(auth_client, monkeypatch):
+    monkeypatch.setattr(app, "vm_exists", lambda name: True)
+    app.set_vm_template("tpl-vm", True)
+    response = auth_client.post("/vm/tpl-vm/start", data={"csrf_token": "test-csrf-token"}, follow_redirects=False)
+    assert response.status_code == 303
+    assert "clone_error" in response.headers["location"]
+
+
+def test_metrics_endpoint(auth_client):
+    response = auth_client.get("/live/metrics")
+    assert response.status_code == 200
+    payload = response.json()
+    assert payload["ok"]
+    assert "series" in payload

tests/test_backup_core.py

@@ -0,0 +1,87 @@
+import time
+from pathlib import Path
+
+import backup_core
+
+
+def test_backup_filename_validation():
+    assert backup_core.valid_backup_filename("vm1_20260815_010101.qcow2")
+    assert backup_core.valid_backup_filename("a.qcow2")
+    assert not backup_core.valid_backup_filename("../evil.qcow2")
+    assert not backup_core.valid_backup_filename("x.img")
+    assert not backup_core.valid_backup_filename("")
+
+
+def test_backup_path_traversal_guard():
+    assert backup_core.backup_path("vm1", "vm1_x.qcow2") is not None
+    assert backup_core.backup_path("vm1", "../../etc/passwd") is None
+    assert backup_core.backup_path("../vm1", "x.qcow2") is None
+    assert backup_core.backup_path("vm1", "nested/x.qcow2") is None
+
+
+def test_new_backup_path_shape():
+    path = backup_core.new_backup_path("vm1")
+    assert path is not None
+    assert path.name.startswith("vm1_")
+    assert path.suffix == ".qcow2"
+    assert backup_core.new_backup_path("bad name") is None
+
+
+def test_rotation_keeps_newest():
+    vm_dir = backup_core.backup_dir_for("vm1")
+    vm_dir.mkdir(parents=True)
+    for index in range(5):
+        item = vm_dir / f"vm1_{index}.qcow2"
+        item.write_text("x")
+        stamp = time.time() - (5 - index) * 60
+        import os
+        os.utime(item, (stamp, stamp))
+    removed = backup_core.rotate_backups("vm1", keep=2)
+    assert len(removed) == 3
+    remaining = sorted(p.name for p in vm_dir.glob("*.qcow2"))
+    assert remaining == ["vm1_3.qcow2", "vm1_4.qcow2"]
+
+
+def test_rotation_ignores_bad_input():
+    assert backup_core.rotate_backups("no-such-vm", keep=3) == []
+    assert backup_core.rotate_backups("vm1", keep=0) == []
+
+
+def test_schedule_roundtrip():
+    config = backup_core.save_schedule(True, ["vm2", "vm1", "bad name", "vm1"], 99)
+    assert config == {"enabled": True, "vms": ["vm1", "vm2"], "keep": 30}
+    assert backup_core.load_schedule() == config
+    config = backup_core.save_schedule(False, [], 0)
+    assert backup_core.load_schedule() == {"enabled": False, "vms": [], "keep": 1}
+
+
+def test_schedule_defaults_when_missing():
+    assert backup_core.load_schedule() == {"enabled": False, "vms": [], "keep": backup_core.DEFAULT_KEEP}
+
+
+def test_backup_script_contents():
+    target = Path("/var/lib/virtuality/backups/vm1/vm1_x.qcow2")
+    script = backup_core.build_backup_script("vm1", "/var/lib/virtuality/images/vm1.qcow2", target)
+    assert "virsh suspend vm1" in script
+    assert "qemu-img convert -p -c -O qcow2" in script
+    assert "virsh resume vm1" in script
+    assert f"{target}.part" in script  # атомарная запись через .part
+    restore = backup_core.build_restore_script("/var/lib/virtuality/images/vm1.qcow2", target)
+    assert "qemu-img convert -p -O qcow2" in restore
+    assert ".restore" in restore
+
+
+def test_first_disk_of_parsing(monkeypatch):
+    sample = """ Type   Device   Target   Source
+------------------------------------------------
+ file   disk     vda      /var/lib/virtuality/images/vm1.qcow2
+ file   cdrom    sda      /var/lib/virtuality/iso/ubuntu.iso
+"""
+
+    class FakeResult:
+        returncode = 0
+        stdout = sample
+        stderr = ""
+
+    monkeypatch.setattr(backup_core, "_run", lambda cmd, timeout=60: FakeResult())
+    assert backup_core.first_disk_of("vm1") == "/var/lib/virtuality/images/vm1.qcow2"

tests/test_cloud_images.py

@@ -0,0 +1,80 @@
+import cloud_images
+
+
+def test_catalog_entries_consistent():
+    keys = [entry["key"] for entry in cloud_images.CATALOG]
+    assert len(keys) == len(set(keys))
+    for entry in cloud_images.CATALOG:
+        assert entry["arch"] in ("x86_64", "aarch64")
+        assert entry["url"].startswith("https://")
+        assert cloud_images.safe_image_filename(entry["filename"])
+
+
+def test_catalog_entry_lookup():
+    assert cloud_images.catalog_entry("ubuntu-24.04-x86_64") is not None
+    assert cloud_images.catalog_entry("nope") is None
+
+
+def test_catalog_for_arch_puts_native_first():
+    entries = cloud_images.catalog_for_arch("aarch64")
+    natives = [e["native"] for e in entries]
+    assert natives == sorted(natives, reverse=True)
+    assert entries[0]["arch"] == "aarch64"
+    entries_x86 = cloud_images.catalog_for_arch("x86_64")
+    assert entries_x86[0]["arch"] == "x86_64"
+
+
+def test_safe_image_filename():
+    assert cloud_images.safe_image_filename("ubuntu-24.04-x86_64.qcow2")
+    assert cloud_images.safe_image_filename("img.img")
+    assert cloud_images.safe_image_filename("../../etc/passwd") is None
+    assert cloud_images.safe_image_filename("../escape.qcow2") == "escape.qcow2"  # basename-санитизация
+    assert cloud_images.safe_image_filename("evil.iso") is None
+
+
+def test_image_path_traversal_guard(tmp_path, monkeypatch):
+    monkeypatch.setattr(cloud_images, "CLOUD_IMAGES_DIR", tmp_path)
+    assert cloud_images.image_path_by_name("ok.qcow2") == tmp_path / "ok.qcow2"
+    # Каталоги отбрасываются до basename — результат всегда внутри CLOUD_IMAGES_DIR
+    assert cloud_images.image_path_by_name("../escape.qcow2") == tmp_path / "escape.qcow2"
+    assert cloud_images.image_path_by_name("bad name.qcow2") is None
+
+
+def test_valid_cloud_username():
+    assert cloud_images.valid_cloud_username("admin")
+    assert cloud_images.valid_cloud_username("_svc-user")
+    assert not cloud_images.valid_cloud_username("Admin")
+    assert not cloud_images.valid_cloud_username("1abc")
+    assert not cloud_images.valid_cloud_username("")
+
+
+def test_valid_ssh_key():
+    assert cloud_images.valid_ssh_key("")  # ключ необязателен
+    assert cloud_images.valid_ssh_key("ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAIExample user@host")
+    assert cloud_images.valid_ssh_key("ssh-rsa AAAAB3NzaC1yc2E=")
+    assert not cloud_images.valid_ssh_key("not a key")
+    assert not cloud_images.valid_ssh_key("ssh-ed25519")
+
+
+def test_build_user_data_with_password():
+    data = cloud_images.build_user_data("vm1", "admin", 'p"ss', "")
+    assert data.startswith("#cloud-config")
+    assert "hostname: vm1" in data
+    assert "name: admin" in data
+    assert "ssh_pwauth: true" in data
+    assert '"p\\"ss"' in data  # пароль экранирован как JSON/YAML-строка
+    assert "ssh_authorized_keys" not in data
+
+
+def test_build_user_data_with_key_only():
+    key = "ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAIExample user@host"
+    data = cloud_images.build_user_data("vm1", "admin", "", key)
+    assert "ssh_pwauth: false" in data
+    assert key in data
+    assert "chpasswd" not in data
+
+
+def test_build_meta_data():
+    meta = cloud_images.build_meta_data("vm1")
+    assert "instance-id: virtuality-vm1" in meta
+    assert "local-hostname: vm1" in meta

tests/test_validators.py

@@ -0,0 +1,93 @@
+import app
+
+
+def test_valid_vm_name():
+    assert app.valid_vm_name("ubuntu-test")
+    assert app.valid_vm_name("vm1.local_x")
+    assert not app.valid_vm_name("")
+    assert not app.valid_vm_name("a")  # минимум 2 символа
+    assert not app.valid_vm_name("-starts-with-dash")
+    assert not app.valid_vm_name("has space")
+    assert not app.valid_vm_name("x" * 64)
+
+
+def test_valid_snapshot_name():
+    assert app.valid_snapshot_name("before-update")
+    assert app.valid_snapshot_name("s1")
+    assert app.valid_snapshot_name("a")  # один символ допустим
+    assert not app.valid_snapshot_name("")
+    assert not app.valid_snapshot_name("bad name")
+    assert not app.valid_snapshot_name(".hidden")
+
+
+def test_safe_iso_filename():
+    assert app.safe_iso_filename("ubuntu-24.04.iso") == "ubuntu-24.04.iso"
+    assert app.safe_iso_filename("my image.iso") == "my-image.iso"
+    assert app.safe_iso_filename("../../etc/passwd") is None
+    assert app.safe_iso_filename("evil.qcow2") is None
+    assert app.safe_iso_filename("") is None
+
+
+def test_iso_path_traversal_guard(tmp_path, monkeypatch):
+    monkeypatch.setattr(app, "ISO_DIR", tmp_path)
+    assert app.iso_path_by_name("ok.iso") == tmp_path / "ok.iso"
+    # Пути с каталогами санитизируются до basename и остаются внутри ISO_DIR
+    assert app.iso_path_by_name("../escape.iso") == tmp_path / "escape.iso"
+    assert app.iso_path_by_name("nested/evil.iso") == tmp_path / "evil.iso"
+    assert app.iso_path_by_name("не-ascii.iso") != tmp_path / "не-ascii.iso"  # кириллица заменяется
+    assert app.iso_path_by_name("no-extension") is None
+
+
+def test_progress_parsers():
+    assert app.wget_progress(10, " 512000K .......... 42% 10.5M 2m30s") == 42
+    assert app.wget_progress(50, "no percent here") == 50
+    assert app.wget_progress(10, "100%") == 99  # качаем — не показываем 100 до mv
+    assert app.qemu_convert_progress(0, "    (42.00/100%)") == 42
+    assert app.qemu_convert_progress(77, "    (42.00/100%)") == 77  # прогресс не откатывается
+    assert app.progress_from_line(10, "Allocating disk...") == 30
+    assert app.progress_from_line(50, "Creating domain...") == 75
+
+
+def test_snapshot_list_parser(monkeypatch):
+    sample = """ Name             Creation Time               State
+--------------------------------------------------------
+ before-update    2026-08-15 12:10:33 +0300   running
+ clean            2026-08-14 09:01:02 +0300   shutoff
+"""
+    monkeypatch.setattr(app, "run_cmd", lambda cmd, timeout=12: {"ok": True, "stdout": sample, "stderr": "", "code": 0, "cmd": ""})
+    snaps = app.list_vm_snapshots("vm1")
+    assert [s["name"] for s in snaps] == ["before-update", "clean"]
+    assert snaps[0]["state"] == "running"
+    assert snaps[0]["created"] == "2026-08-15 12:10:33 +0300"
+
+
+def test_snapshot_action_rejects_bad_name():
+    ok, message = app.snapshot_action("vm1", "create", "bad name")
+    assert not ok
+    assert "Имя снапшота" in message
+
+
+def test_vm_templates_flag(tmp_path, monkeypatch):
+    monkeypatch.setattr(app, "VM_TEMPLATES_FILE", tmp_path / "vm_templates.json")
+    assert app.load_vm_templates() == []
+    app.set_vm_template("vm1", True)
+    app.set_vm_template("vm2", True)
+    assert app.is_vm_template("vm1")
+    assert sorted(app.load_vm_templates()) == ["vm1", "vm2"]
+    app.set_vm_template("vm1", False)
+    assert not app.is_vm_template("vm1")
+
+
+def test_ttl_cache(monkeypatch):
+    calls = []
+
+    def producer():
+        calls.append(1)
+        return len(calls)
+
+    app.cache_invalidate()
+    assert app.cached("k1", 60, producer) == 1
+    assert app.cached("k1", 60, producer) == 1
+    assert len(calls) == 1
+    app.cache_invalidate("k1")
+    assert app.cached("k1", 60, producer) == 2