@@ -348,10 +472,19 @@
SSH Config Template
+
+
{% endblock %}
{% block extra_js %}
-
+
+
+
{% endblock %}
diff --git a/tests/integration/test_instance_deletion.py b/tests/integration/test_instance_deletion.py
new file mode 100644
index 0000000..8e1b78c
--- /dev/null
+++ b/tests/integration/test_instance_deletion.py
@@ -0,0 +1,101 @@
+"""
+Integration tests for instance deletion with jump user cleanup.
+
+These tests verify that when an instance is deleted, the associated
+jump user is also properly cleaned up.
+
+Note: We test the service layer directly rather than through HTTP endpoints
+to avoid complex mocking of the entire request pipeline.
+"""
+import pytest
+from unittest.mock import patch, MagicMock, call
+
+
+class TestJumpUserCleanupOnDeletion:
+ """Test that jump users are properly deleted when instances are deleted."""
+
+ @patch("services.jump_user_service.delete_jump_user")
+ def test_delete_jump_user_called_with_instance_name(self, mock_delete_jump_user):
+ """Verify delete_jump_user is called with the instance name."""
+ from services.jump_user_service import delete_jump_user
+
+ mock_delete_jump_user.return_value = {"success": True}
+
+ # Call the function directly
+ result = delete_jump_user("test-vm-001")
+
+ assert result["success"] is True
+ mock_delete_jump_user.assert_called_once_with("test-vm-001")
+
+ @patch("services.jump_user_service.subprocess.run")
+ @patch("services.jump_user_service.jump_user_exists")
+ def test_jump_user_deleted_with_correct_username(
+ self, mock_exists, mock_run
+ ):
+ """Verify that userdel is called with the instance name, not a generic username."""
+ from services.jump_user_service import delete_jump_user
+
+ mock_exists.return_value = True
+ mock_run.return_value = MagicMock(returncode=0)
+
+ # Delete jump user with instance name
+ delete_jump_user("my-custom-vm")
+
+ # Verify userdel was called with instance name
+ userdel_call = mock_run.call_args_list[0]
+ userdel_args = userdel_call[0][0]
+
+ # Should contain instance name, not 'ubuntu' or 'root'
+ assert "my-custom-vm" in userdel_args
+ assert "ubuntu" not in userdel_args
+
+ # Verify -r flag is present (remove home directory)
+ assert "-r" in userdel_args
+
+ @patch("services.jump_user_service.subprocess.run")
+ @patch("services.jump_user_service.jump_user_exists")
+ def test_multiple_instances_have_separate_jump_users(
+ self, mock_exists, mock_run
+ ):
+ """Verify that each instance has its own jump user that can be deleted independently."""
+ from services.jump_user_service import create_jump_user, delete_jump_user
+
+ mock_exists.return_value = False # User doesn't exist initially
+ mock_run.return_value = MagicMock(returncode=0)
+
+ instance_names = ["vm-001", "vm-002", "vm-003"]
+
+ # Create jump users for each instance
+ for name in instance_names:
+ result = create_jump_user(
+ username="ubuntu", # This should be ignored
+ public_key="ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAI test@example.com",
+ instance_name=name,
+ )
+ assert result["success"] is True
+
+ # Verify useradd was called 3 times with different names
+ useradd_calls = [c for c in mock_run.call_args_list if "useradd" in c[0][0]]
+ assert len(useradd_calls) == 3
+
+ # Reset mock to test deletion
+ mock_run.reset_mock()
+ mock_exists.return_value = True
+
+ # Delete each jump user
+ for name in instance_names:
+ result = delete_jump_user(name)
+ assert result["success"] is True
+
+ # Verify userdel was called 3 times with correct names
+ userdel_calls = [c for c in mock_run.call_args_list if "userdel" in c[0][0]]
+ assert len(userdel_calls) == 3
+
+ # Verify each call used the correct instance name
+ deleted_names = []
+ for call_item in userdel_calls:
+ args = call_item[0][0]
+ deleted_names.append(args[-1]) # Last arg is the username
+
+ assert set(deleted_names) == set(instance_names)
+
diff --git a/tests/services/test_jump_user_service.py b/tests/services/test_jump_user_service.py
new file mode 100644
index 0000000..77dca5e
--- /dev/null
+++ b/tests/services/test_jump_user_service.py
@@ -0,0 +1,210 @@
+"""
+Tests for Jump User Service.
+
+These tests verify that jump users are properly created and deleted,
+ensuring each instance has its own unique jump user.
+"""
+import pytest
+from unittest.mock import patch, MagicMock, call
+from services.jump_user_service import (
+ create_jump_user,
+ delete_jump_user,
+ jump_user_exists,
+ update_jump_user_keys,
+)
+
+
+class TestJumpUserCreation:
+ """Test jump user creation functionality."""
+
+ @patch("services.jump_user_service.subprocess.run")
+ @patch("services.jump_user_service.jump_user_exists")
+ def test_create_jump_user_with_instance_name(
+ self, mock_exists, mock_run
+ ):
+ """Test that jump user is created with instance name as username."""
+ mock_exists.return_value = False
+ mock_run.return_value = MagicMock(returncode=0)
+
+ result = create_jump_user(
+ username="ubuntu", # This should be ignored when instance_name is provided
+ public_key="ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAI test@example.com",
+ instance_name="vm-test-001",
+ )
+
+ assert result["success"] is True
+ assert result["created"] is True
+
+ # Verify useradd was called with instance name, not 'ubuntu'
+ useradd_call = mock_run.call_args_list[0]
+ assert "vm-test-001" in useradd_call[0][0]
+ assert "ubuntu" not in useradd_call[0][0]
+
+ @patch("services.jump_user_service.jump_user_exists")
+ def test_create_jump_user_fails_if_user_exists(self, mock_exists):
+ """Test that creation fails if user already exists (no reuse)."""
+ mock_exists.return_value = True
+
+ result = create_jump_user(
+ username="vm-test-001",
+ public_key="ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAI test@example.com",
+ instance_name="vm-test-001",
+ )
+
+ assert result["success"] is False
+ assert "already exists" in result["message"]
+
+ @patch("services.jump_user_service.subprocess.run")
+ @patch("services.jump_user_service.jump_user_exists")
+ def test_create_jump_user_creates_ssh_directory(
+ self, mock_exists, mock_run
+ ):
+ """Test that .ssh directory is created for the jump user."""
+ mock_exists.return_value = False
+ mock_run.return_value = MagicMock(returncode=0)
+
+ create_jump_user(
+ username="test-user",
+ public_key="ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAI test@example.com",
+ instance_name="vm-test-001",
+ )
+
+ # Check that mkdir was called for .ssh directory
+ mkdir_calls = [
+ call for call in mock_run.call_args_list if "mkdir" in call[0][0]
+ ]
+ assert len(mkdir_calls) > 0
+
+ # Verify the path uses the instance name
+ mkdir_call = mkdir_calls[0]
+ assert "vm-test-001/.ssh" in mkdir_call[0][0][3]
+
+ @patch("services.jump_user_service.subprocess.run")
+ @patch("services.jump_user_service.jump_user_exists")
+ def test_create_jump_user_sets_correct_permissions(
+ self, mock_exists, mock_run
+ ):
+ """Test that correct permissions are set for SSH files."""
+ mock_exists.return_value = False
+ mock_run.return_value = MagicMock(returncode=0)
+
+ create_jump_user(
+ username="test-user",
+ public_key="ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAI test@example.com",
+ instance_name="vm-test-001",
+ )
+
+ # Check chmod calls for .ssh directory (700) and authorized_keys (600)
+ chmod_calls = [
+ call for call in mock_run.call_args_list if "chmod" in call[0][0]
+ ]
+
+ # Should have at least 2 chmod calls (700 for .ssh, 600 for authorized_keys)
+ assert len(chmod_calls) >= 2
+
+ # Verify permissions
+ chmod_700 = [c for c in chmod_calls if "700" in c[0][0]]
+ chmod_600 = [c for c in chmod_calls if "600" in c[0][0]]
+
+ assert len(chmod_700) > 0
+ assert len(chmod_600) > 0
+
+
+class TestJumpUserDeletion:
+ """Test jump user deletion functionality."""
+
+ @patch("services.jump_user_service.subprocess.run")
+ @patch("services.jump_user_service.jump_user_exists")
+ def test_delete_jump_user_with_instance_name(self, mock_exists, mock_run):
+ """Test that jump user is deleted using instance name."""
+ mock_exists.return_value = True
+ mock_run.return_value = MagicMock(returncode=0)
+
+ result = delete_jump_user("vm-test-001")
+
+ assert result["success"] is True
+
+ # Verify userdel was called with instance name
+ userdel_call = mock_run.call_args_list[0]
+ assert "vm-test-001" in userdel_call[0][0]
+
+ @patch("services.jump_user_service.jump_user_exists")
+ def test_delete_jump_user_when_user_does_not_exist(self, mock_exists):
+ """Test that deletion succeeds silently if user doesn't exist."""
+ mock_exists.return_value = False
+
+ result = delete_jump_user("vm-test-001")
+
+ assert result["success"] is True
+ assert "does not exist" in result["message"]
+
+ @patch("services.jump_user_service.subprocess.run")
+ @patch("services.jump_user_service.jump_user_exists")
+ def test_delete_jump_user_removes_home_directory(
+ self, mock_exists, mock_run
+ ):
+ """Test that userdel is called with -r flag to remove home directory."""
+ mock_exists.return_value = True
+ mock_run.return_value = MagicMock(returncode=0)
+
+ delete_jump_user("vm-test-001")
+
+ # Verify userdel was called with -r flag
+ userdel_call = mock_run.call_args_list[0]
+ assert "-r" in userdel_call[0][0]
+
+
+class TestJumpUserIntegration:
+ """Integration tests for jump user lifecycle."""
+
+ @patch("services.jump_user_service.subprocess.run")
+ @patch("services.jump_user_service.jump_user_exists")
+ def test_create_and_delete_same_instance_name(self, mock_exists, mock_run):
+ """Test that a jump user can be created and then deleted."""
+ # First, user doesn't exist
+ mock_exists.side_effect = [False, True] # Create: False, Delete: True
+ mock_run.return_value = MagicMock(returncode=0)
+
+ # Create user
+ create_result = create_jump_user(
+ username="test",
+ public_key="ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAI test@example.com",
+ instance_name="vm-lifecycle-test",
+ )
+ assert create_result["success"] is True
+
+ # Delete user
+ delete_result = delete_jump_user("vm-lifecycle-test")
+ assert delete_result["success"] is True
+
+ @patch("services.jump_user_service.subprocess.run")
+ @patch("services.jump_user_service.jump_user_exists")
+ def test_multiple_instances_have_unique_users(self, mock_exists, mock_run):
+ """Test that multiple instances create separate jump users."""
+ mock_exists.return_value = False
+ mock_run.return_value = MagicMock(returncode=0)
+
+ instance_names = ["vm-001", "vm-002", "vm-003"]
+
+ for name in instance_names:
+ result = create_jump_user(
+ username="ubuntu", # Same fallback username
+ public_key="ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAI test@example.com",
+ instance_name=name,
+ )
+ assert result["success"] is True
+
+ # Verify useradd was called 3 times with different names
+ useradd_calls = [
+ call for call in mock_run.call_args_list if "useradd" in call[0][0]
+ ]
+ assert len(useradd_calls) == 3
+
+ # Verify each call used a different instance name
+ called_names = []
+ for call_item in useradd_calls:
+ args = call_item[0][0]
+ # Extract username from args (it's the last argument)
+ called_names.append(args[-1])
+
+ assert set(called_names) == set(instance_names)
diff --git a/tests/utils/test_virtualization.py b/tests/utils/test_virtualization.py
new file mode 100644
index 0000000..ba55e33
--- /dev/null
+++ b/tests/utils/test_virtualization.py
@@ -0,0 +1,209 @@
+"""
+Tests for virtualization detection utility.
+"""
+import pytest
+from unittest.mock import patch, MagicMock, mock_open
+from utils.virtualization import (
+ is_virtualization_supported,
+ get_virtualization_info,
+ check_lxd_vm_support,
+)
+
+
+class TestVirtualizationDetection:
+ """Test virtualization detection functions."""
+
+ @patch("utils.virtualization.os.path.exists")
+ def test_kvm_device_exists(self, mock_exists):
+ """Test that KVM device detection works."""
+ mock_exists.return_value = True
+
+ # Clear cache before test
+ is_virtualization_supported.cache_clear()
+
+ result = is_virtualization_supported()
+
+ assert result is True
+ mock_exists.assert_called_with("/dev/kvm")
+
+ @patch("utils.virtualization.os.path.exists")
+ @patch("builtins.open", new_callable=mock_open, read_data="flags : fpu vmx mmx sse")
+ def test_intel_vmx_detected(self, mock_file, mock_exists):
+ """Test that Intel VT-x (vmx) flag is detected."""
+ mock_exists.return_value = False
+
+ # Clear cache before test
+ is_virtualization_supported.cache_clear()
+ get_virtualization_info.cache_clear()
+
+ result = is_virtualization_supported()
+
+ assert result is True
+
+ @patch("utils.virtualization.os.path.exists")
+ @patch("builtins.open", new_callable=mock_open, read_data="flags : fpu svm mmx sse")
+ def test_amd_svm_detected(self, mock_file, mock_exists):
+ """Test that AMD-V (svm) flag is detected."""
+ mock_exists.return_value = False
+
+ # Clear cache before test
+ is_virtualization_supported.cache_clear()
+ get_virtualization_info.cache_clear()
+
+ result = is_virtualization_supported()
+
+ assert result is True
+
+ @patch("utils.virtualization.os.path.exists")
+ @patch("builtins.open", new_callable=mock_open, read_data="flags : fpu mmx sse")
+ def test_no_virtualization_flags(self, mock_file, mock_exists):
+ """Test that missing virtualization flags returns False."""
+ mock_exists.return_value = False
+
+ # Clear cache before test
+ is_virtualization_supported.cache_clear()
+ get_virtualization_info.cache_clear()
+
+ result = is_virtualization_supported()
+
+ assert result is False
+
+ @patch("utils.virtualization.os.path.exists")
+ @patch("utils.virtualization.subprocess.run")
+ @patch("builtins.open", new_callable=mock_open, read_data="flags : fpu mmx sse")
+ def test_kvm_ok_command_success(self, mock_file, mock_run, mock_exists):
+ """Test that kvm-ok command success indicates support."""
+ mock_exists.return_value = False
+ mock_run.return_value = MagicMock(returncode=0, stdout="KVM acceleration can be used")
+
+ # Clear cache before test
+ is_virtualization_supported.cache_clear()
+ get_virtualization_info.cache_clear()
+
+ result = is_virtualization_supported()
+
+ assert result is True
+
+ @patch("utils.virtualization.os.path.exists")
+ @patch("utils.virtualization.subprocess.run")
+ @patch("builtins.open", new_callable=mock_open, read_data="flags : fpu mmx sse")
+ def test_kvm_ok_command_failure(self, mock_file, mock_run, mock_exists):
+ """Test that kvm-ok command failure returns False."""
+ mock_exists.return_value = False
+ mock_run.return_value = MagicMock(returncode=1, stdout="KVM is not available")
+
+ # Clear cache before test
+ is_virtualization_supported.cache_clear()
+ get_virtualization_info.cache_clear()
+
+ result = is_virtualization_supported()
+
+ assert result is False
+
+ @patch("utils.virtualization.os.path.exists")
+ @patch("utils.virtualization.subprocess.run")
+ @patch("builtins.open", new_callable=mock_open, read_data="flags : fpu mmx sse")
+ def test_kvm_ok_not_available(self, mock_file, mock_run, mock_exists):
+ """Test that missing kvm-ok command doesn't crash."""
+ mock_exists.return_value = False
+ mock_run.side_effect = FileNotFoundError("kvm-ok not found")
+
+ # Clear cache before test
+ is_virtualization_supported.cache_clear()
+ get_virtualization_info.cache_clear()
+
+ result = is_virtualization_supported()
+
+ assert result is False
+
+
+class TestGetVirtualizationInfo:
+ """Test get_virtualization_info function."""
+
+ @patch("utils.virtualization.os.path.exists")
+ @patch("builtins.open", new_callable=mock_open, read_data="flags : fpu vmx mmx sse")
+ def test_info_with_intel_cpu(self, mock_file, mock_exists):
+ """Test virtualization info for Intel CPU."""
+ mock_exists.return_value = True
+
+ # Clear cache before test
+ get_virtualization_info.cache_clear()
+
+ info = get_virtualization_info()
+
+ assert info["supported"] is True
+ assert info["kvm_device"] is True
+ assert info["cpu_vendor"] == "Intel"
+ assert "vmx" in info["cpu_flags"]
+
+ @patch("utils.virtualization.os.path.exists")
+ @patch("builtins.open", new_callable=mock_open, read_data="flags : fpu svm mmx sse")
+ def test_info_with_amd_cpu(self, mock_file, mock_exists):
+ """Test virtualization info for AMD CPU."""
+ mock_exists.return_value = False
+
+ # Clear cache before test
+ get_virtualization_info.cache_clear()
+
+ info = get_virtualization_info()
+
+ assert info["supported"] is True
+ assert info["kvm_device"] is False
+ assert info["cpu_vendor"] == "AMD"
+ assert "svm" in info["cpu_flags"]
+
+ @patch("utils.virtualization.os.path.exists")
+ @patch("builtins.open", new_callable=mock_open, read_data="flags : fpu mmx sse")
+ def test_info_without_virtualization(self, mock_file, mock_exists):
+ """Test virtualization info when not supported."""
+ mock_exists.return_value = False
+
+ # Clear cache before test
+ get_virtualization_info.cache_clear()
+
+ info = get_virtualization_info()
+
+ assert info["supported"] is False
+ assert info["kvm_device"] is False
+ assert info["cpu_vendor"] is None
+ assert "vmx" not in info["cpu_flags"]
+ assert "svm" not in info["cpu_flags"]
+
+
+class TestLXDVMSupport:
+ """Test check_lxd_vm_support function."""
+
+ @patch("utils.virtualization.get_virtualization_info")
+ def test_vm_supported(self, mock_get_info):
+ """Test when VMs are supported."""
+ mock_get_info.return_value = {
+ "supported": True,
+ "kvm_device": True,
+ "cpu_flags": ["vmx"],
+ "cpu_vendor": "Intel",
+ "message": "Hardware virtualization is available"
+ }
+
+ result = check_lxd_vm_support()
+
+ assert result["vm_supported"] is True
+ assert result["container_supported"] is True
+ assert result["recommendation"] == ""
+
+ @patch("utils.virtualization.get_virtualization_info")
+ def test_vm_not_supported(self, mock_get_info):
+ """Test when VMs are not supported."""
+ mock_get_info.return_value = {
+ "supported": False,
+ "kvm_device": False,
+ "cpu_flags": [],
+ "cpu_vendor": None,
+ "message": "Hardware virtualization is not available"
+ }
+
+ result = check_lxd_vm_support()
+
+ assert result["vm_supported"] is False
+ assert result["container_supported"] is True
+ assert "recommendation" in result
+ assert len(result["recommendation"]) > 0
diff --git a/utils/virtualization.py b/utils/virtualization.py
new file mode 100644
index 0000000..3ab58d8
--- /dev/null
+++ b/utils/virtualization.py
@@ -0,0 +1,145 @@
+"""System utility functions for hardware and virtualization detection"""
+import os
+import subprocess
+from functools import lru_cache
+
+
+@lru_cache(maxsize=1)
+def is_virtualization_supported() -> bool:
+ """
+ Check if the system supports hardware virtualization (KVM/QEMU).
+
+ This is required for creating LXD virtual machines (VMs).
+ Containers do not require virtualization support.
+
+ Checks performed (in order):
+ 1. Check if /dev/kvm exists (KVM device)
+ 2. Check CPU flags for vmx (Intel) or svm (AMD)
+ 3. Try to run kvm-ok command if available
+
+ Returns:
+ True if virtualization is supported, False otherwise
+ """
+ # Check 1: KVM device exists
+ if os.path.exists("/dev/kvm"):
+ return True
+
+ # Check 2: CPU flags contain virtualization extensions
+ try:
+ with open("/proc/cpuinfo", "r") as f:
+ cpuinfo = f.read().lower()
+ # Intel VT-x or AMD-V
+ if "vmx" in cpuinfo or "svm" in cpuinfo:
+ return True
+ except (FileNotFoundError, PermissionError):
+ pass
+
+ # Check 3: Try kvm-ok command (if available)
+ try:
+ result = subprocess.run(
+ ["kvm-ok"],
+ capture_output=True,
+ text=True,
+ timeout=5
+ )
+ if result.returncode == 0:
+ return True
+ except (FileNotFoundError, subprocess.TimeoutExpired):
+ pass
+
+ return False
+
+
+@lru_cache(maxsize=1)
+def get_virtualization_info() -> dict:
+ """
+ Get detailed information about virtualization support.
+
+ Returns:
+ dict with virtualization support details
+ """
+ info = {
+ "supported": False,
+ "kvm_device": False,
+ "cpu_flags": [],
+ "cpu_vendor": None,
+ "kvm_ok_available": False,
+ "message": ""
+ }
+
+ # Check KVM device
+ info["kvm_device"] = os.path.exists("/dev/kvm")
+
+ # Check CPU flags
+ try:
+ with open("/proc/cpuinfo", "r") as f:
+ cpuinfo = f.read()
+ # Extract flags from first CPU
+ for line in cpuinfo.split("\n"):
+ if line.startswith("flags") or line.startswith("Features"):
+ flags = line.split(":", 1)[1].strip().lower().split()
+ info["cpu_flags"] = flags
+ if "vmx" in flags:
+ info["cpu_vendor"] = "Intel"
+ elif "svm" in flags:
+ info["cpu_vendor"] = "AMD"
+ break
+ except (FileNotFoundError, PermissionError):
+ pass
+
+ # Check kvm-ok
+ try:
+ result = subprocess.run(
+ ["kvm-ok"],
+ capture_output=True,
+ text=True,
+ timeout=5
+ )
+ info["kvm_ok_available"] = True
+ if result.returncode == 0:
+ info["message"] = "KVM acceleration can be used"
+ except (FileNotFoundError, subprocess.TimeoutExpired):
+ info["kvm_ok_available"] = False
+
+ # Determine overall support
+ info["supported"] = info["kvm_device"] or \
+ "vmx" in info["cpu_flags"] or \
+ "svm" in info["cpu_flags"]
+
+ # Generate message
+ if info["supported"]:
+ vendor = info["cpu_vendor"] or "Unknown"
+ info["message"] = f"Hardware virtualization ({vendor}) is available"
+ else:
+ info["message"] = "Hardware virtualization is not available on this system"
+
+ return info
+
+
+def check_lxd_vm_support() -> dict:
+ """
+ Check if LXD can create virtual machines on this system.
+
+ This combines virtualization detection with LXD-specific checks.
+
+ Returns:
+ dict with LXD VM support details
+ """
+ virt_info = get_virtualization_info()
+
+ result = {
+ "vm_supported": virt_info["supported"],
+ "container_supported": True, # Containers always supported
+ "details": virt_info,
+ "recommendation": ""
+ }
+
+ if not virt_info["supported"]:
+ result["recommendation"] = (
+ "Virtual machines require hardware virtualization (Intel VT-x or AMD-V). "
+ "This system does not appear to have virtualization support enabled. "
+ "You can still create containers, which do not require virtualization. "
+ "If this is a cloud VM, check if your provider offers VMs with nested virtualization enabled."
+ )
+
+ return result