Merge pull request #9 from kpn/feature/credentialsecrets-model

NEW - Added credential secret

katka/admin.py

@@ -1,13 +1,30 @@
 from django.contrib import admin
 
 from katka.fields import username_on_model
-from katka.models import Team
+from katka.models import Credential, CredentialSecret, Project, Team
 
 
-@admin.register(Team)
-class TeamAdmin(admin.ModelAdmin):
-    fields = ('name', 'group')
-
+class WithUsernameAdminModel(admin.ModelAdmin):
     def save_model(self, request, obj, form, change):
         with username_on_model(self.model, request.user.username):
             super().save_model(request, obj, form, change)
+
+
+@admin.register(Team)
+class TeamAdmin(WithUsernameAdminModel):
+    fields = ('name', 'group')
+
+
+@admin.register(Project)
+class ProjectAdmin(WithUsernameAdminModel):
+    fields = ('name', 'slug', 'team')
+
+
+@admin.register(Credential)
+class CredentialAdmin(WithUsernameAdminModel):
+    fields = ('name', 'slug', 'credential_type', 'team')
+
+
+@admin.register(CredentialSecret)
+class CredentialSecretAdmin(WithUsernameAdminModel):
+    fields = ('key', 'value', 'credential')

katka/migrations/0004_credential_secret.py

@@ -0,0 +1,35 @@
+# Generated by Django 2.1.5 on 2019-02-12 02:12
+
+import django.db.models.deletion
+from django.db import migrations, models
+
+import encrypted_model_fields.fields
+import katka.fields
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('katka', '0003_credential_and_unique_slugs'),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name='CredentialSecret',
+            fields=[
+                ('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+                ('created', models.DateTimeField(auto_now_add=True)),
+                ('created_username', katka.fields.AutoUsernameField(max_length=50)),
+                ('modified', models.DateTimeField(auto_now=True)),
+                ('modified_username', katka.fields.AutoUsernameField(max_length=50)),
+                ('status', models.CharField(choices=[('active', 'active'), ('inactive', 'inactive')], default='active', max_length=50)),
+                ('key', models.CharField(max_length=50)),
+                ('value', encrypted_model_fields.fields.EncryptedCharField()),
+                ('credential', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to='katka.Credential')),
+            ],
+        ),
+        migrations.AlterUniqueTogether(
+            name='credentialsecret',
+            unique_together={('credential', 'key')},
+        ),
+    ]

katka/models.py

@@ -3,6 +3,7 @@
 from django.contrib.auth.models import Group
 from django.db import models
 
+from encrypted_model_fields.fields import EncryptedCharField
 from katka.auditedmodel import AuditedModel
 from katka.fields import KatkaSlugField
 
@@ -42,3 +43,15 @@ class Meta:
 
     def __str__(self):  # pragma: no cover
         return f'{self.name}'
+
+
+class CredentialSecret(AuditedModel):
+    key = models.CharField(max_length=50)
+    value = EncryptedCharField(max_length=200)
+    credential = models.ForeignKey(Credential, on_delete=models.CASCADE)
+
+    class Meta:
+        unique_together = ('credential', 'key')
+
+    def __str__(self):  # pragma: no cover
+        return f'{self.credential.name}/{self.key}'

katka/serializers.py

@@ -1,6 +1,6 @@
 from django.contrib.auth.models import Group
 
-from katka.models import Credential, Project, Team
+from katka.models import Credential, CredentialSecret, Project, Team
 from rest_framework import serializers
 from rest_framework.exceptions import NotFound, PermissionDenied
 
@@ -78,3 +78,22 @@ class CredentialSerializer(TeamChildMixin, serializers.ModelSerializer):
     class Meta:
         model = Credential
         fields = ('name', 'slug', 'team')
+
+
+class CredentialSecretSerializer(serializers.ModelSerializer):
+    class Meta:
+        model = CredentialSecret
+        fields = ('key', 'value', 'credential')
+
+    def to_internal_value(self, data):
+        data['credential'] = self.context['view'].kwargs['credential_pk']
+        return super().to_internal_value(data)
+
+    def validate(self, attrs):
+        team_pui = self.context['view'].kwargs['team_public_identifier']
+        if not self.context['request'].user.groups.filter(team__public_identifier=team_pui).exists():
+            if Team.objects.filter(public_identifier=team_pui).exists():
+                raise PermissionDenied('User is not a member of this group')
+            raise NotFound
+
+        return attrs

katka/urls.py

@@ -10,17 +10,20 @@
 
 router = routers.SimpleRouter()
 router.register('team', views.TeamViewSet, basename='team')
-router.register('project', views.ProjectViewSet, basename='project')
 
 project_router = NestedSimpleRouter(router, 'team', lookup='team')
 project_router.register('project', views.ProjectViewSet, basename='project')
 
 credential_router = NestedSimpleRouter(router, 'team', lookup='team')
 credential_router.register('credential', views.CredentialViewSet, basename='credential')
 
+secrets_router = NestedSimpleRouter(credential_router, 'credential', lookup='credential')
+secrets_router.register('secrets', views.CredentialSecretsViewSet, basename='secrets')
+
 urlpatterns = [
     path('admin/', admin.site.urls),
     path('', include(router.urls)),
     path('', include(project_router.urls)),
     path('', include(credential_router.urls)),
+    path('', include(secrets_router.urls)),
 ]

katka/views.py

@@ -1,5 +1,5 @@
-from katka.models import Credential, Project, Team
-from katka.serializers import CredentialSerializer, ProjectSerializer, TeamSerializer
+from katka.models import Credential, CredentialSecret, Project, Team
+from katka.serializers import CredentialSecretSerializer, CredentialSerializer, ProjectSerializer, TeamSerializer
 from katka.viewsets import AuditViewSet
 
 
@@ -31,3 +31,13 @@ class CredentialViewSet(AuditViewSet):
     def get_queryset(self):
         user_groups = self.request.user.groups.all()
         return super().get_queryset().filter(team__group__in=user_groups)
+
+
+class CredentialSecretsViewSet(AuditViewSet):
+    model = CredentialSecret
+    serializer_class = CredentialSecretSerializer
+    lookup_field = 'key'
+
+    def get_queryset(self):
+        user_groups = self.request.user.groups.all()
+        return super().get_queryset().filter(credential__team__group__in=user_groups)

tests/integration/conftest.py

@@ -58,6 +58,24 @@ def credential(team):
     return credential
 
 
+@pytest.fixture
+def secret(credential):
+    secret = models.CredentialSecret(key='access_token', value='full_access_value', credential=credential)
+    with username_on_model(models.CredentialSecret, 'initial'):
+        secret.save()
+
+    return secret
+
+
+@pytest.fixture
+def deactivated_secret(secret):
+    secret.status = 'inactive'
+    with username_on_model(models.CredentialSecret, 'initial'):
+        secret.save()
+
+    return secret
+
+
 @pytest.fixture
 def user(group):
     u = User.objects.create_user('test_user', None, None)

tests/integration/test_credential_secret_view.py

@@ -0,0 +1,122 @@
+from uuid import UUID
+
+import pytest
+from katka import models
+from katka.constants import STATUS_INACTIVE
+
+
+@pytest.mark.django_db
+class TestCredentialSecretViewSetUnauthenticated:
+    """
+    When a user is not logged in, no group information is available, so nothing is returned.
+
+    For listing, that would be an empty list for other operations, an error like the object could
+    not be found, except on create (you need to be part of a group and anonymous users do not have any)
+    """
+
+    def test_list(self, client, team, credential, secret):
+        url = f'/team/{team.public_identifier}/credential/{credential.public_identifier}/secrets/'
+        response = client.get(url)
+        assert response.status_code == 200
+        parsed = response.json()
+        assert len(parsed) == 0
+
+    def test_list_unknown_team(self, client, team, credential, secret):
+        unknown_pui = '00000000-0000-0000-0000-000000000000'
+        url = f'/team/{unknown_pui}/credential/{credential.public_identifier}/secrets/'
+        response = client.get(url)
+        assert response.status_code == 200
+        parsed = response.json()
+        assert len(parsed) == 0
+
+    def test_get(self, client, team, credential, secret):
+        url = f'/team/{team.public_identifier}/credential/{credential.public_identifier}/secrets/{secret.key}/'
+        response = client.get(url)
+        assert response.status_code == 404
+
+    def test_delete(self, client, team, credential, secret):
+        url = f'/team/{team.public_identifier}/credential/{credential.public_identifier}/secrets/{secret.key}/'
+        response = client.delete(url)
+        assert response.status_code == 404
+
+    def test_update(self, client, team, credential, secret):
+        url = f'/team/{team.public_identifier}/credential/{credential.public_identifier}/secrets/{secret.key}/'
+        data = {'name': 'B-Team', 'group': 'group1'}
+        response = client.put(url, data, content_type='application/json')
+        assert response.status_code == 404
+
+    def test_partial_update(self, client, team, credential, secret):
+        url = f'/team/{team.public_identifier}/credential/{credential.public_identifier}/secrets/{secret.key}/'
+        response = client.patch(url, {'name': 'B-Team'}, content_type='application/json')
+        assert response.status_code == 404
+
+    def test_create(self, client, team, credential):
+        url = f'/team/{team.public_identifier}/credential/{credential.public_identifier}/secrets/'
+        data = {'key': 'access_token', 'value': 'my_secret_access_token'}
+        response = client.post(url, data, content_type='application/json')
+        assert response.status_code == 403
+
+
+@pytest.mark.django_db
+class TestCredentialViewSet:
+    def test_list(self, client, logged_in_user, team, credential, secret):
+        response = client.get(f'/team/{team.public_identifier}/credential/{credential.public_identifier}/secrets/')
+        assert response.status_code == 200
+        parsed = response.json()
+        assert len(parsed) == 1
+        secret = parsed[0]
+        assert secret['key'] == 'access_token'
+        assert secret['value'] == 'full_access_value'
+        assert UUID(secret['credential']) == credential.public_identifier
+
+    def test_get(self, client, logged_in_user, team, credential, secret):
+        url = f'/team/{team.public_identifier}/credential/{credential.public_identifier}/secrets/{secret.key}/'
+        response = client.get(url)
+        assert response.status_code == 200
+        secret = response.json()
+        assert secret['key'] == 'access_token'
+        assert secret['value'] == 'full_access_value'
+        assert UUID(secret['credential']) == credential.public_identifier
+
+    def test_get_excludes_inactive(self, client, logged_in_user, team, credential, deactivated_secret):
+        url = f'/team/{team.public_identifier}/credential/{credential.public_identifier}/' + \
+              f'secrets/{deactivated_secret.key}/'
+        response = client.get(url)
+        assert response.status_code == 404
+
+    def test_delete(self, client, logged_in_user, team, credential, secret):
+        url = f'/team/{team.public_identifier}/credential/{credential.public_identifier}/secrets/{secret.key}/'
+        response = client.delete(url)
+        assert response.status_code == 204
+        s = models.CredentialSecret.objects.get(key=secret.key, credential=credential)
+        assert s.status == STATUS_INACTIVE
+
+    def test_update(self, client, logged_in_user, team, credential, secret):
+        url = f'/team/{team.public_identifier}/credential/{credential.public_identifier}/secrets/{secret.key}/'
+        data = {'key': 'access_token', 'value': 'new value'}
+        response = client.put(url, data, content_type='application/json')
+        assert response.status_code == 200
+        s = models.CredentialSecret.objects.get(key=secret.key, credential=credential)
+        assert s.value == 'new value'
+
+    def test_update_nonexistent_team(self, client, logged_in_user, team, credential, secret):
+        unknown_pui = '00000000-0000-0000-0000-000000000000'
+        url = f'/team/{unknown_pui}/credential/{credential.public_identifier}/secrets/{secret.key}/'
+        data = {'key': 'access_token', 'value': 'new value'}
+        response = client.put(url, data, content_type='application/json')
+        assert response.status_code == 404
+
+    def test_partial_update(self, client, logged_in_user, team, credential, secret):
+        url = f'/team/{team.public_identifier}/credential/{credential.public_identifier}/secrets/{secret.key}/'
+        data = {'value': 'new value'}
+        response = client.patch(url, data, content_type='application/json')
+        assert response.status_code == 200
+        s = models.CredentialSecret.objects.get(key=secret.key, credential=credential)
+        assert s.value == 'new value'
+
+    def test_create(self, client, logged_in_user, team, credential, secret):
+        url = f'/team/{team.public_identifier}/credential/{credential.public_identifier}/secrets/'
+        response = client.post(url, {'key': 'password', 'value': 'new value'}, content_type='application/json')
+        assert response.status_code == 201
+        models.CredentialSecret.objects.get(key='access_token', credential=credential)
+        assert models.CredentialSecret.objects.count() == 2

tests/unit/test_admin.py

@@ -3,8 +3,9 @@
 from django.test.client import RequestFactory
 
 import pytest
-from katka.admin import TeamAdmin
-from katka.models import Team
+from katka.admin import CredentialAdmin, CredentialSecretAdmin, ProjectAdmin, TeamAdmin
+from katka.fields import username_on_model
+from katka.models import Credential, CredentialSecret, Project, Team
 
 
 @pytest.fixture
@@ -22,6 +23,22 @@ def group():
     return g
 
 
+@pytest.fixture
+def team(group):
+    t = Team(name='team', group=group)
+    with username_on_model(Team, 'audit_user'):
+        t.save()
+    return t
+
+
+@pytest.fixture
+def credential(team):
+    c = Credential(name='team', team=team)
+    with username_on_model(Credential, 'audit_user'):
+        c.save()
+    return c
+
+
 @pytest.mark.django_db
 class TestTeamAdmin:
     def test_save_stores_username(self, mock_request, group):
@@ -30,3 +47,36 @@ def test_save_stores_username(self, mock_request, group):
         t.save_model(mock_request, obj, None, None)
         assert obj.created_username == 'mock1'
         assert obj.modified_username == 'mock1'
+
+
+@pytest.mark.django_db
+class TestProjectAdmin:
+    def test_save_stores_username(self, mock_request, team):
+        p = ProjectAdmin(Project, AdminSite())
+        obj = Project(name='Project D', slug='PRJD', team=team)
+        p.save_model(mock_request, obj, None, None)
+
+        assert obj.created_username == 'mock1'
+        assert obj.modified_username == 'mock1'
+
+
+@pytest.mark.django_db
+class TestCredentialAdmin:
+    def test_save_stores_username(self, mock_request, team):
+        c = CredentialAdmin(Credential, AdminSite())
+        obj = Credential(name='Credential D', slug='CRED', team=team)
+        c.save_model(mock_request, obj, None, None)
+
+        assert obj.created_username == 'mock1'
+        assert obj.modified_username == 'mock1'
+
+
+@pytest.mark.django_db
+class TestCredentialSecretAdmin:
+    def test_save_stores_username(self, mock_request, credential):
+        c = CredentialSecretAdmin(CredentialSecret, AdminSite())
+        obj = CredentialSecret(key='access_key', value='supersecret', credential=credential)
+        c.save_model(mock_request, obj, None, None)
+
+        assert obj.created_username == 'mock1'
+        assert obj.modified_username == 'mock1'