(feature) Generate filesystem-backed SSH credentials
Add filesystem-backed SSH credentials for the native systemd deployment path. Generated keys are stored below POBSYNC_HOME with 0600 permissions, while Django keeps the public key, fingerprint, path, and selection metadata. Add a Django SSH key generation view, delete action for unused generated keys, and a management command used by the installer to ensure a default backup key exists. Update runtime config to use generated key paths directly as IdentityFile, extend host checks to verify key readability, and keep legacy uploaded keys available for compatibility.
This commit is contained in:
@@ -115,3 +115,21 @@ class DjangoConfigSourceTests(TestCase):
|
||||
self.assertFalse(global_identity_file.exists())
|
||||
|
||||
self.assertIn(f"-oIdentityFile={host_identity_file}", cfg["ssh"]["options"])
|
||||
|
||||
def test_filesystem_ssh_credential_uses_existing_key_path(self) -> None:
|
||||
with TemporaryDirectory() as tmp:
|
||||
identity_file = Path(tmp) / "identity"
|
||||
identity_file.write_text("PRIVATE KEY\n", encoding="utf-8")
|
||||
identity_file.chmod(0o600)
|
||||
credential = SshCredential.objects.create(name="backup-key", key_path=str(identity_file), public_key="PUBLIC")
|
||||
GlobalConfig.objects.create(
|
||||
name="default",
|
||||
backup_root="/backups",
|
||||
pobsync_home="/opt/pobsync",
|
||||
default_ssh_credential=credential,
|
||||
)
|
||||
HostConfig.objects.create(host="web-01", address="web-01.example.test")
|
||||
|
||||
cfg = DjangoConfigSource().effective_config_for_host("web-01")
|
||||
|
||||
self.assertIn(f"-oIdentityFile={identity_file}", cfg["ssh"]["options"])
|
||||
|
||||
@@ -5,9 +5,11 @@ from pathlib import Path
|
||||
from tempfile import TemporaryDirectory
|
||||
|
||||
from django import forms
|
||||
from django.test import SimpleTestCase
|
||||
from django.core.management import call_command
|
||||
from django.test import SimpleTestCase, TestCase, override_settings
|
||||
|
||||
from pobsync_backend.forms import normalize_private_key, validate_ssh_private_key
|
||||
from pobsync_backend.models import GlobalConfig, SshCredential
|
||||
|
||||
|
||||
class SshCredentialValidationTests(SimpleTestCase):
|
||||
@@ -38,3 +40,23 @@ class SshCredentialValidationTests(SimpleTestCase):
|
||||
validate_ssh_private_key("-----BEGIN RSA PRIVATE KEY-----\nabc\n-----END RSA PRIVATE KEY-----")
|
||||
|
||||
self.assertIn("PEM private keys are not supported", str(exc.exception))
|
||||
|
||||
|
||||
class SshCredentialManagementTests(TestCase):
|
||||
def test_ensure_ssh_key_command_generates_default_key(self) -> None:
|
||||
with TemporaryDirectory() as tmp, override_settings(POBSYNC_HOME=str(Path(tmp) / "home")):
|
||||
call_command("ensure_pobsync_ssh_key", "--name", "default")
|
||||
|
||||
credential = SshCredential.objects.get(name="default")
|
||||
self.assertTrue(credential.generated)
|
||||
self.assertTrue(Path(credential.key_path).exists())
|
||||
self.assertTrue(credential.public_key.startswith("ssh-ed25519 "))
|
||||
|
||||
def test_ensure_ssh_key_command_sets_global_default_when_available(self) -> None:
|
||||
global_config = GlobalConfig.objects.create(name="default", backup_root="/backups")
|
||||
|
||||
with TemporaryDirectory() as tmp, override_settings(POBSYNC_HOME=str(Path(tmp) / "home")):
|
||||
call_command("ensure_pobsync_ssh_key", "--name", "default", "--set-global-default")
|
||||
|
||||
global_config.refresh_from_db()
|
||||
self.assertEqual(global_config.default_ssh_credential.name, "default")
|
||||
|
||||
@@ -163,6 +163,45 @@ class ViewTests(TestCase):
|
||||
self.assertEqual(credential.private_key, "UPLOADED PRIVATE KEY\n")
|
||||
self.assertEqual(credential.public_key, "DERIVED PUBLIC KEY")
|
||||
|
||||
def test_ssh_credentials_view_generates_filesystem_key(self) -> None:
|
||||
self.client.force_login(self.staff_user)
|
||||
with TemporaryDirectory() as tmp, override_settings(POBSYNC_HOME=str(Path(tmp) / "home")):
|
||||
response = self.client.post(
|
||||
reverse("generate_ssh_credential"),
|
||||
{
|
||||
"name": "generated-key",
|
||||
"key_type": "ed25519",
|
||||
"set_global_default": "",
|
||||
"known_hosts": "",
|
||||
"notes": "generated",
|
||||
},
|
||||
follow=True,
|
||||
)
|
||||
|
||||
self.assertRedirects(response, reverse("ssh_credentials"))
|
||||
self.assertContains(response, "SSH key generated for generated-key.")
|
||||
credential = SshCredential.objects.get(name="generated-key")
|
||||
self.assertTrue(credential.generated)
|
||||
self.assertEqual(credential.private_key, "")
|
||||
self.assertTrue(credential.public_key.startswith("ssh-ed25519 "))
|
||||
self.assertTrue(Path(credential.key_path).exists())
|
||||
|
||||
def test_ssh_credentials_view_deletes_unused_generated_key(self) -> None:
|
||||
self.client.force_login(self.staff_user)
|
||||
with TemporaryDirectory() as tmp, override_settings(POBSYNC_HOME=str(Path(tmp) / "home")):
|
||||
credential = SshCredential.objects.create(name="generated-key")
|
||||
from pobsync_backend.ssh_keys import generate_ssh_key
|
||||
|
||||
generate_ssh_key(credential)
|
||||
key_path = Path(credential.key_path)
|
||||
|
||||
response = self.client.post(reverse("delete_ssh_credential", args=[credential.id]), follow=True)
|
||||
|
||||
self.assertRedirects(response, reverse("ssh_credentials"))
|
||||
self.assertContains(response, "SSH key deleted: generated-key.")
|
||||
self.assertFalse(SshCredential.objects.exists())
|
||||
self.assertFalse(key_path.exists())
|
||||
|
||||
def test_ssh_credentials_view_rejects_invalid_key(self) -> None:
|
||||
self.client.force_login(self.staff_user)
|
||||
|
||||
@@ -284,6 +323,15 @@ class ViewTests(TestCase):
|
||||
self.assertEqual(config.retention_daily, 7)
|
||||
self.assertEqual(config.retention_yearly, 1)
|
||||
|
||||
def test_global_config_form_defaults_to_first_generated_key(self) -> None:
|
||||
self.client.force_login(self.staff_user)
|
||||
credential = SshCredential.objects.create(name="default", key_path="/var/lib/pobsync/state/ssh-credentials/1/identity")
|
||||
|
||||
response = self.client.get(reverse("edit_global_config"))
|
||||
|
||||
self.assertEqual(response.status_code, 200)
|
||||
self.assertContains(response, f'value="{credential.id}" selected')
|
||||
|
||||
def test_global_config_form_renders_static_container_backup_root_on_edit(self) -> None:
|
||||
self.client.force_login(self.staff_user)
|
||||
GlobalConfig.objects.create(
|
||||
|
||||
Reference in New Issue
Block a user