399 lines
16 KiB
Python
399 lines
16 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from datetime import timedelta
|
|
from types import SimpleNamespace
|
|
from unittest.mock import patch
|
|
|
|
from django.contrib.auth import get_user_model
|
|
from django.contrib.auth.models import Permission
|
|
from django.test import TestCase
|
|
from django.utils import timezone
|
|
from guardian.shortcuts import assign_perm
|
|
from ninja.errors import HttpError
|
|
|
|
from apps.keys.models import SSHCertificate, SSHKey
|
|
from keywarden.api.routers import keys as keys_router
|
|
|
|
|
|
class KeysRouterHelperTests(TestCase):
|
|
def setUp(self) -> None:
|
|
super().setUp()
|
|
User = get_user_model()
|
|
self.user = User.objects.create_user(
|
|
username="keys-helper",
|
|
email="keys-helper@example.com",
|
|
password="pass12345",
|
|
)
|
|
|
|
def _create_key(self, *, owner=None, name: str = "helper-key", key_b64: str = "AAAA") -> SSHKey:
|
|
key = SSHKey(user=owner or self.user, name=name)
|
|
key.set_public_key(f"ssh-ed25519 {key_b64} comment")
|
|
key.save()
|
|
return key
|
|
|
|
def test_ensure_certificate_raises_when_key_is_revoked(self) -> None:
|
|
key = self._create_key()
|
|
key.is_active = False
|
|
key.save(update_fields=["is_active"])
|
|
|
|
with self.assertRaises(HttpError):
|
|
keys_router._ensure_certificate(key, self.user)
|
|
|
|
def test_ensure_certificate_issues_when_missing_or_expired(self) -> None:
|
|
key_missing = self._create_key(name="missing")
|
|
key_expired = self._create_key(name="expired", key_b64="AAAB")
|
|
SSHCertificate.objects.create(
|
|
key=key_expired,
|
|
user=self.user,
|
|
certificate="ssh-ed25519-cert-v01 AAAA",
|
|
serial=1001,
|
|
principals=["keys-helper"],
|
|
valid_after=timezone.now() - timedelta(days=2),
|
|
valid_before=timezone.now() - timedelta(days=1),
|
|
is_active=True,
|
|
)
|
|
issued = SimpleNamespace(serial=999, certificate="ssh-ed25519-cert-v01 BBBB")
|
|
with patch("keywarden.api.routers.keys.issue_certificate_for_key", return_value=issued) as mocked_issue:
|
|
self.assertIs(keys_router._ensure_certificate(key_missing, self.user), issued)
|
|
self.assertIs(keys_router._ensure_certificate(key_expired, self.user), issued)
|
|
self.assertEqual(mocked_issue.call_count, 2)
|
|
|
|
def test_ensure_certificate_returns_existing_active_cert(self) -> None:
|
|
key = self._create_key()
|
|
cert = SSHCertificate.objects.create(
|
|
key=key,
|
|
user=self.user,
|
|
certificate="ssh-ed25519-cert-v01 AAAA",
|
|
serial=42,
|
|
principals=["keys-helper"],
|
|
valid_after=timezone.now() - timedelta(minutes=5),
|
|
valid_before=timezone.now() + timedelta(days=1),
|
|
is_active=True,
|
|
)
|
|
|
|
with patch("keywarden.api.routers.keys.issue_certificate_for_key") as mocked_issue:
|
|
result = keys_router._ensure_certificate(key, self.user)
|
|
self.assertEqual(result.id, cert.id)
|
|
mocked_issue.assert_not_called()
|
|
|
|
|
|
class KeysApiRouterTests(TestCase):
|
|
def setUp(self) -> None:
|
|
super().setUp()
|
|
User = get_user_model()
|
|
self.user = User.objects.create_user(
|
|
username="keys-user",
|
|
email="keys-user@example.com",
|
|
password="pass12345",
|
|
)
|
|
self.other_user = User.objects.create_user(
|
|
username="keys-other",
|
|
email="keys-other@example.com",
|
|
password="pass12345",
|
|
)
|
|
|
|
def _grant_global_perm(self, user, codename: str) -> None:
|
|
perm = Permission.objects.get(content_type__app_label="keys", codename=codename)
|
|
user.user_permissions.add(perm)
|
|
|
|
def _create_key(self, *, owner=None, name: str = "api-key", key_b64: str = "AAAA") -> SSHKey:
|
|
key = SSHKey(user=owner or self.user, name=name)
|
|
key.set_public_key(f"ssh-ed25519 {key_b64} comment")
|
|
key.save()
|
|
return key
|
|
|
|
def _dummy_cert(self):
|
|
now = timezone.now()
|
|
return SimpleNamespace(
|
|
serial=1234,
|
|
valid_after=now,
|
|
valid_before=now + timedelta(days=1),
|
|
principals=[self.user.username],
|
|
certificate="ssh-ed25519-cert-v01 AAAATEST",
|
|
)
|
|
|
|
def test_list_keys_returns_only_object_permitted_keys(self) -> None:
|
|
visible = self._create_key(owner=self.other_user, name="visible")
|
|
hidden = self._create_key(owner=self.other_user, name="hidden", key_b64="AAAB")
|
|
assign_perm("view_sshkey", self.user, visible)
|
|
self.client.force_login(self.user)
|
|
|
|
response = self.client.get("/api/v1/keys/")
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
ids = {item["id"] for item in response.json()}
|
|
self.assertIn(visible.id, ids)
|
|
self.assertNotIn(hidden.id, ids)
|
|
|
|
def test_list_keys_honors_global_filter_by_user(self) -> None:
|
|
self._grant_global_perm(self.user, "view_sshkey")
|
|
other_key = self._create_key(owner=self.other_user, name="other", key_b64="AAAC")
|
|
self._create_key(owner=self.user, name="self", key_b64="AAAD")
|
|
self.client.force_login(self.user)
|
|
|
|
response = self.client.get(f"/api/v1/keys/?user_id={self.other_user.id}&limit=20&offset=0")
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
payload = response.json()
|
|
self.assertEqual(len(payload), 1)
|
|
self.assertEqual(payload[0]["id"], other_key.id)
|
|
|
|
def test_create_key_requires_add_permission(self) -> None:
|
|
self.client.force_login(self.user)
|
|
|
|
response = self.client.post(
|
|
"/api/v1/keys/",
|
|
data=json.dumps({"name": "laptop", "public_key": "ssh-ed25519 AAAA"}),
|
|
content_type="application/json",
|
|
)
|
|
|
|
self.assertEqual(response.status_code, 403)
|
|
|
|
def test_create_key_for_self_success(self) -> None:
|
|
self._grant_global_perm(self.user, "add_sshkey")
|
|
self.client.force_login(self.user)
|
|
with patch("keywarden.api.routers.keys.issue_certificate_for_key", return_value=self._dummy_cert()) as mocked:
|
|
response = self.client.post(
|
|
"/api/v1/keys/",
|
|
data=json.dumps({"name": "laptop", "public_key": "ssh-ed25519 AAAA"}),
|
|
content_type="application/json",
|
|
)
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
payload = response.json()
|
|
self.assertEqual(payload["user_id"], self.user.id)
|
|
self.assertEqual(payload["name"], "laptop")
|
|
self.assertTrue(SSHKey.objects.filter(id=payload["id"]).exists())
|
|
mocked.assert_called_once()
|
|
|
|
def test_create_key_for_other_user_requires_admin_view_permission(self) -> None:
|
|
self._grant_global_perm(self.user, "add_sshkey")
|
|
self.client.force_login(self.user)
|
|
|
|
response = self.client.post(
|
|
"/api/v1/keys/",
|
|
data=json.dumps(
|
|
{"name": "other-user-key", "public_key": "ssh-ed25519 AAAA", "user_id": self.other_user.id}
|
|
),
|
|
content_type="application/json",
|
|
)
|
|
|
|
self.assertEqual(response.status_code, 403)
|
|
|
|
def test_create_key_for_other_user_with_admin_perms(self) -> None:
|
|
self._grant_global_perm(self.user, "add_sshkey")
|
|
self._grant_global_perm(self.user, "view_sshkey")
|
|
self.client.force_login(self.user)
|
|
with patch("keywarden.api.routers.keys.issue_certificate_for_key", return_value=self._dummy_cert()):
|
|
response = self.client.post(
|
|
"/api/v1/keys/",
|
|
data=json.dumps(
|
|
{"name": "other-user-key", "public_key": "ssh-ed25519 AAAA", "user_id": self.other_user.id}
|
|
),
|
|
content_type="application/json",
|
|
)
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
payload = response.json()
|
|
self.assertEqual(payload["user_id"], self.other_user.id)
|
|
|
|
def test_create_key_unknown_user_returns_404(self) -> None:
|
|
self._grant_global_perm(self.user, "add_sshkey")
|
|
self._grant_global_perm(self.user, "view_sshkey")
|
|
self.client.force_login(self.user)
|
|
|
|
response = self.client.post(
|
|
"/api/v1/keys/",
|
|
data=json.dumps(
|
|
{"name": "other-user-key", "public_key": "ssh-ed25519 AAAA", "user_id": 999999}
|
|
),
|
|
content_type="application/json",
|
|
)
|
|
|
|
self.assertEqual(response.status_code, 404)
|
|
|
|
def test_create_key_validation_paths_raise_type_error(self) -> None:
|
|
self._grant_global_perm(self.user, "add_sshkey")
|
|
self.client.force_login(self.user)
|
|
|
|
with self.assertRaises(TypeError):
|
|
self.client.post(
|
|
"/api/v1/keys/",
|
|
data=json.dumps({"name": " ", "public_key": "ssh-ed25519 AAAA"}),
|
|
content_type="application/json",
|
|
)
|
|
with self.assertRaises(TypeError):
|
|
self.client.post(
|
|
"/api/v1/keys/",
|
|
data=json.dumps({"name": "badkey", "public_key": "not-a-key"}),
|
|
content_type="application/json",
|
|
)
|
|
|
|
def test_create_key_duplicate_fingerprint_raises_type_error(self) -> None:
|
|
self._grant_global_perm(self.user, "add_sshkey")
|
|
self.client.force_login(self.user)
|
|
self._create_key(owner=self.user, name="existing")
|
|
|
|
with self.assertRaises(TypeError):
|
|
self.client.post(
|
|
"/api/v1/keys/",
|
|
data=json.dumps({"name": "duplicate", "public_key": "ssh-ed25519 AAAA"}),
|
|
content_type="application/json",
|
|
)
|
|
|
|
def test_get_key_not_found_and_forbidden(self) -> None:
|
|
self.client.force_login(self.user)
|
|
missing = self.client.get("/api/v1/keys/999999")
|
|
self.assertEqual(missing.status_code, 404)
|
|
|
|
key = self._create_key(owner=self.other_user, name="forbidden")
|
|
forbidden = self.client.get(f"/api/v1/keys/{key.id}")
|
|
self.assertEqual(forbidden.status_code, 403)
|
|
|
|
def test_get_key_success(self) -> None:
|
|
key = self._create_key(name="visible-key")
|
|
assign_perm("view_sshkey", self.user, key)
|
|
self.client.force_login(self.user)
|
|
|
|
response = self.client.get(f"/api/v1/keys/{key.id}")
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(response.json()["id"], key.id)
|
|
|
|
def test_issue_certificate_requires_object_permission(self) -> None:
|
|
key = self._create_key(owner=self.other_user, name="cert-forbidden")
|
|
self.client.force_login(self.user)
|
|
|
|
forbidden = self.client.post(f"/api/v1/keys/{key.id}/certificate")
|
|
self.assertEqual(forbidden.status_code, 403)
|
|
|
|
missing = self.client.post("/api/v1/keys/999999/certificate")
|
|
self.assertEqual(missing.status_code, 404)
|
|
|
|
def test_issue_certificate_success(self) -> None:
|
|
key = self._create_key(name="cert-ok")
|
|
assign_perm("view_sshkey", self.user, key)
|
|
self.client.force_login(self.user)
|
|
cert = self._dummy_cert()
|
|
with patch("keywarden.api.routers.keys.issue_certificate_for_key", return_value=cert) as mocked:
|
|
response = self.client.post(f"/api/v1/keys/{key.id}/certificate")
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(response.json()["serial"], cert.serial)
|
|
mocked.assert_called_once_with(key, created_by=self.user)
|
|
|
|
def test_download_certificate_and_hash(self) -> None:
|
|
key = self._create_key(name="download")
|
|
assign_perm("view_sshkey", self.user, key)
|
|
self.client.force_login(self.user)
|
|
cert = self._dummy_cert()
|
|
|
|
with patch("keywarden.api.routers.keys._ensure_certificate", return_value=cert):
|
|
cert_response = self.client.get(f"/api/v1/keys/{key.id}/certificate")
|
|
hash_response = self.client.get(f"/api/v1/keys/{key.id}/certificate.sha256")
|
|
|
|
self.assertEqual(cert_response.status_code, 200)
|
|
self.assertIn("attachment;", cert_response["Content-Disposition"])
|
|
self.assertIn("AAAATEST", cert_response.content.decode("utf-8"))
|
|
|
|
self.assertEqual(hash_response.status_code, 200)
|
|
hash_payload = hash_response.content.decode("utf-8")
|
|
self.assertIn("keywarden-", hash_payload)
|
|
self.assertTrue(hash_payload.endswith(".pub\n"))
|
|
|
|
def test_download_certificate_requires_permission(self) -> None:
|
|
key = self._create_key(owner=self.other_user, name="download-forbidden")
|
|
self.client.force_login(self.user)
|
|
|
|
cert_response = self.client.get(f"/api/v1/keys/{key.id}/certificate")
|
|
hash_response = self.client.get(f"/api/v1/keys/{key.id}/certificate.sha256")
|
|
|
|
self.assertEqual(cert_response.status_code, 403)
|
|
self.assertEqual(hash_response.status_code, 403)
|
|
|
|
def test_update_key_paths(self) -> None:
|
|
key = self._create_key(owner=self.other_user, name="editable", key_b64="AAAE")
|
|
self.client.force_login(self.user)
|
|
|
|
forbidden = self.client.patch(
|
|
f"/api/v1/keys/{key.id}",
|
|
data=json.dumps({"name": "renamed"}),
|
|
content_type="application/json",
|
|
)
|
|
self.assertEqual(forbidden.status_code, 403)
|
|
|
|
self._grant_global_perm(self.user, "change_sshkey")
|
|
assign_perm("change_sshkey", self.user, key)
|
|
|
|
renamed = self.client.patch(
|
|
f"/api/v1/keys/{key.id}",
|
|
data=json.dumps({"name": "renamed"}),
|
|
content_type="application/json",
|
|
)
|
|
self.assertEqual(renamed.status_code, 200)
|
|
key.refresh_from_db()
|
|
self.assertEqual(key.name, "renamed")
|
|
|
|
with patch("keywarden.api.routers.keys.revoke_certificate_for_key") as mocked_revoke:
|
|
revoked = self.client.patch(
|
|
f"/api/v1/keys/{key.id}",
|
|
data=json.dumps({"is_active": False}),
|
|
content_type="application/json",
|
|
)
|
|
self.assertEqual(revoked.status_code, 200)
|
|
key.refresh_from_db()
|
|
self.assertFalse(key.is_active)
|
|
self.assertIsNotNone(key.revoked_at)
|
|
mocked_revoke.assert_called_once_with(key)
|
|
|
|
with patch("keywarden.api.routers.keys.issue_certificate_for_key", return_value=self._dummy_cert()) as mocked_issue:
|
|
reactivated = self.client.patch(
|
|
f"/api/v1/keys/{key.id}",
|
|
data=json.dumps({"is_active": True}),
|
|
content_type="application/json",
|
|
)
|
|
self.assertEqual(reactivated.status_code, 200)
|
|
key.refresh_from_db()
|
|
self.assertTrue(key.is_active)
|
|
self.assertIsNone(key.revoked_at)
|
|
mocked_issue.assert_called_once_with(key, created_by=self.user)
|
|
|
|
with self.assertRaises(TypeError):
|
|
self.client.patch(
|
|
f"/api/v1/keys/{key.id}",
|
|
data=json.dumps({}),
|
|
content_type="application/json",
|
|
)
|
|
|
|
not_found = self.client.patch(
|
|
"/api/v1/keys/999999",
|
|
data=json.dumps({"name": "x"}),
|
|
content_type="application/json",
|
|
)
|
|
self.assertEqual(not_found.status_code, 404)
|
|
|
|
def test_delete_key_paths(self) -> None:
|
|
key = self._create_key(owner=self.other_user, name="delete-me", key_b64="AAAF")
|
|
self.client.force_login(self.user)
|
|
|
|
forbidden = self.client.delete(f"/api/v1/keys/{key.id}")
|
|
self.assertEqual(forbidden.status_code, 403)
|
|
|
|
self._grant_global_perm(self.user, "delete_sshkey")
|
|
assign_perm("delete_sshkey", self.user, key)
|
|
|
|
with patch("keywarden.api.routers.keys.revoke_certificate_for_key") as mocked_revoke:
|
|
deleted = self.client.delete(f"/api/v1/keys/{key.id}")
|
|
self.assertEqual(deleted.status_code, 204)
|
|
key.refresh_from_db()
|
|
self.assertFalse(key.is_active)
|
|
self.assertIsNotNone(key.revoked_at)
|
|
mocked_revoke.assert_called_once_with(key)
|
|
|
|
second_delete = self.client.delete(f"/api/v1/keys/{key.id}")
|
|
self.assertEqual(second_delete.status_code, 204)
|
|
|
|
missing = self.client.delete("/api/v1/keys/999999")
|
|
self.assertEqual(missing.status_code, 404)
|