diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..a794f7b --- /dev/null +++ b/.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 diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..1d4e050 --- /dev/null +++ b/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 diff --git a/tests/test_app_security.py b/tests/test_app_security.py new file mode 100644 index 0000000..0564c60 --- /dev/null +++ b/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 diff --git a/tests/test_backup_core.py b/tests/test_backup_core.py new file mode 100644 index 0000000..8e222ad --- /dev/null +++ b/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" diff --git a/tests/test_cloud_images.py b/tests/test_cloud_images.py new file mode 100644 index 0000000..38f4188 --- /dev/null +++ b/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 diff --git a/tests/test_validators.py b/tests/test_validators.py new file mode 100644 index 0000000..9351c29 --- /dev/null +++ b/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