feat(perms): refine the permission model

Author: duanhongyi

rootfs/api/consumers.py

@@ -20,7 +20,7 @@
 from channels.generic.websocket import AsyncWebsocketConsumer
 
 from .models.app import App
-from .permissions import has_object_permission
+from .permissions import has_app_permission
 
 
 class BaseAppConsumer(AsyncWebsocketConsumer):
@@ -35,7 +35,7 @@ def has_perm(self):
         if permission is None:
             try:
                 app = App.objects.get(id=self.id)
-                permission = has_object_permission(self.scope["user"], app, "GET")
+                permission = has_app_permission(self.scope["user"], app, "GET")
                 if permission[0]:
                     cache.set(key, permission, timeout=self.timeout)
             except App.DoesNotExist:

rootfs/api/migrations/0010_alter_certificate_options.py

@@ -14,7 +14,7 @@ def migration_dryccfile(apps, schema_editor):
 class Migration(migrations.Migration):
 
     dependencies = [
-        ('api', '0011_merge_20240815_0955'),
+        ('api', '0010_appsettings_autorollback'),
     ]
 
     operations = [

…igrations/0012_appsettings_autodeploy.py …igrations/0011_appsettings_autodeploy.pyrootfs/api/migrations/0012_appsettings_autodeploy.py renamed to rootfs/api/migrations/0011_appsettings_autodeploy.py rootfs/api/migrations/0012_appsettings_autodeploy.py renamed to rootfs/api/migrations/0011_appsettings_autodeploy.py

@@ -0,0 +1,67 @@
+# Generated by Django 4.2.15 on 2024-08-30 00:53
+
+import api.utils
+from django.db import migrations, models
+import django.db.models.deletion
+from guardian.shortcuts import assign_perm, get_users_with_perms, remove_perm
+from api.models.app import App, VIEW_APP_PERMISSION, CHANGE_APP_PERMISSION
+from api.models.domain import Domain
+
+
+def migration_permission(apps, schema_editor):
+    for app in App.objects.all():
+        for user in get_users_with_perms(app):
+            remove_perm('use_app', user, app)
+            assign_perm(VIEW_APP_PERMISSION.codename, user, app)
+            assign_perm(CHANGE_APP_PERMISSION.codename, user, app)
+
+
+def migration_certificate(apps, schema_editor):
+    for domain in Domain.objects.all():
+        if domain.certificate:
+            if domain.certificate.app == None:
+                domain.certificate.app = domain.app
+                domain.certificate.save()
+            else:
+                certificate = domain.certificate
+                certificate.pk = None
+                certificate.app = domain.app
+                certificate.save()
+                domain.certificate = certificate
+                domain.save()
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('api', '0011_appsettings_autodeploy'),
+    ]
+
+    operations = [
+        migrations.AlterModelOptions(
+            name='app',
+            options={'ordering': ['id'], 'verbose_name': 'Application'},
+        ),
+        migrations.AddField(
+            model_name='certificate',
+            name='app',
+            field=models.ForeignKey(default=None, on_delete=django.db.models.deletion.CASCADE, to='api.app'),
+            preserve_default=False,
+        ),
+        migrations.AddField(
+            model_name='release',
+            name='conditions',
+            field=models.JSONField(default=list),
+        ),
+        migrations.AlterField(
+            model_name='certificate',
+            name='name',
+            field=models.CharField(max_length=253, validators=[api.utils.validate_label]),
+        ),
+        migrations.AlterUniqueTogether(
+            name='certificate',
+            unique_together={('app', 'name')},
+        ),
+        migrations.RunPython(migration_permission),
+        migrations.RunPython(migration_certificate),
+    ]

rootfs/api/migrations/0011_merge_20240815_0955.py

@@ -10,7 +10,7 @@
 import socket
 from contextlib import closing
 from urllib.parse import urljoin
-from collections import OrderedDict
+from collections import OrderedDict, namedtuple
 from datetime import datetime, timezone
 
 from docker import auth as docker_auth
@@ -31,12 +31,51 @@
 from .tls import TLS
 from .appsettings import AppSettings
 from .volume import Volume
-from .base import UuidAuditedModel, PROCFILE_TYPE_WEB, PROCFILE_TYPE_RUN, DEFAULT_HTTP_PORT, \
-    ObjectPolicy, object_policy_registry
+from .base import UuidAuditedModel, PROCFILE_TYPE_WEB, PROCFILE_TYPE_RUN, DEFAULT_HTTP_PORT
 
 User = get_user_model()
 logger = logging.getLogger(__name__)
-app_policy = ObjectPolicy('id', 'use_app', 'Can use app')  # query field, policy, description
+AppPermission = namedtuple('AppPermission', ['shortname', 'codename', 'description'])
+VIEW_APP_PERMISSION = AppPermission("view", "view_app", "can view app")
+CHANGE_APP_PERMISSION = AppPermission("change", "change_app", "can change app")
+DELETE_APP_PERMISSION = AppPermission("delete", "delete_app", "can delete app")
+
+
+class AppPermissionRegistry(object):
+
+    def __init__(self):
+        self.tags = {}
+        self.permissions = set()
+
+    def get(self, q):
+        permissions = [
+            permission for permission in self.permissions
+            if q == permission.shortname or q == permission.codename
+        ]
+        permissions.extend([
+            permission for permission in self.permissions
+            if permission in self.tags and q in self.tags[permission]
+        ])
+        return permissions[0] if permissions else None
+
+    def register(self, permission, tags=None):
+        if tags:
+            self.tags[permission] = tags
+        self.permissions.add(permission)
+
+    @property
+    def codenames(self):
+        return [permission[1] for permission in self.permissions]
+
+    @property
+    def shortnames(self):
+        return [permission[0] for permission in self.permissions]
+
+
+app_permission_registry = AppPermissionRegistry()
+app_permission_registry.register(VIEW_APP_PERMISSION, ["GET", "HEAD", "OPTION"])
+app_permission_registry.register(CHANGE_APP_PERMISSION, ["POST", "PUT", "PATCH"])
+app_permission_registry.register(DELETE_APP_PERMISSION, ["DELETE"])
 
 
 # http://kubernetes.io/v1.1/docs/design/identifiers.html
@@ -79,7 +118,6 @@ class App(UuidAuditedModel):
 
     class Meta:
         verbose_name = 'Application'
-        permissions = (app_policy[1:], )
         ordering = ['id']
 
     def save(self, *args, **kwargs):
@@ -1168,7 +1206,3 @@ def _gather_app_settings(self, release, app_settings, procfile_type, replicas, v
             'pod_security_context': limit_plan.pod_security_context,
             'container_security_context': limit_plan.container_security_context,
         }
-
-
-# Register policy
-object_policy_registry.register(App, app_policy)

rootfs/api/migrations/0012_alter_app_options_certificate_app_release_conditions_and_more.py

@@ -4,7 +4,6 @@
 import importlib
 from datetime import timedelta
 from functools import partial
-from collections import namedtuple
 from django.db import models
 from django.conf import settings
 from django.utils import timezone
@@ -39,38 +38,6 @@
 def get_anonymous_user_instance(user): return user(id=-1, username=settings.ANONYMOUS_USER_NAME)
 
 
-ObjectPolicy = namedtuple('ObjectPolicy', ['unique', 'codename', 'description'])
-
-
-class ObjectPolicyRegistry(object):
-
-    def __init__(self):
-        self.object_permission_policy_table = dict[type, ObjectPolicy]()
-
-    def all(self) -> list[type, ObjectPolicy]:
-        return self.object_permission_policy_table.items()
-
-    def get(self, query) -> tuple[type, ObjectPolicy]:
-        if isinstance(query, str):
-            for key, value in self.object_permission_policy_table.items():
-                if value.codename == query:
-                    return key, value
-        elif isinstance(query, models.Model):
-            query = type(query)
-            value = self.object_permission_policy_table.get(query)
-            return query if value else None, value
-        elif isinstance(query, type):
-            value = self.object_permission_policy_table.get(query)
-            return query if value else None, value
-        return None, None
-
-    def register(self, cls: type, object_policy: ObjectPolicy) -> None:
-        self.object_permission_policy_table[cls] = object_policy
-
-
-object_policy_registry = ObjectPolicyRegistry()
-
-
 class AuditedModel(models.Model):
     """Add created and updated fields to a model."""
 

rootfs/api/models/app.py

@@ -15,12 +15,11 @@
 from api.utils import validate_label
 from api.exceptions import AlreadyExists, ServiceUnavailable
 from scheduler import KubeException
-from .base import AuditedModel, ObjectPolicy, object_policy_registry
+from .base import AuditedModel
 from .domain import Domain
 
 User = get_user_model()
 logger = logging.getLogger(__name__)
-cert_policy = ObjectPolicy('name', 'use_cert', 'Can use cert')
 
 
 # Note: This is a slightly bug-fixed version of same from ndg-httpsclient.
@@ -104,7 +103,8 @@ class Certificate(AuditedModel):
     Public and private key pair used to secure application traffic at the router.
     """
     owner = models.ForeignKey(User, on_delete=models.PROTECT)
-    name = models.CharField(max_length=253, unique=True, validators=[validate_label])
+    app = models.ForeignKey('App', on_delete=models.CASCADE)
+    name = models.CharField(max_length=253, validators=[validate_label])
     # there is no upper limit on the size of an x.509 certificate
     certificate = models.TextField(validators=[validate_certificate])
     key = models.TextField(validators=[validate_private_key])
@@ -121,8 +121,8 @@ class Certificate(AuditedModel):
     subject = models.TextField(editable=False)
 
     class Meta:
-        permissions = (cert_policy[1:], )
         ordering = ['name', 'common_name', 'expires']
+        unique_together = ('app', 'name')
 
     @property
     def domains(self):
@@ -240,7 +240,3 @@ def detach(self, *args, **kwargs):
                 raise ServiceUnavailable(
                     "Could not delete certificate secret {} for application {}".format(
                         self.certname, namespace)) from e
-
-
-# Register policy
-object_policy_registry.register(Certificate, cert_policy)

rootfs/api/models/base.py

@@ -5,7 +5,6 @@
 from api import manager
 from api.models import app
 from api.models import blocklist
-from api.models.base import object_policy_registry
 
 
 def get_app_status(app):
@@ -19,20 +18,20 @@ def get_app_status(app):
     return True, None
 
 
-def has_object_permission(user, obj, method):
+def has_app_permission(user, obj, method):
     obj = getattr(obj, 'app', obj)
-    object_policy = object_policy_registry.get(obj)[1]
     has_permission, message = False, f"{obj} object does not exist or does not have permission."
     if user.is_superuser:
         has_permission, message = True, None
     elif getattr(obj, "owner", None) == user:
         has_permission, message = True, None
-    elif user.is_staff or (object_policy and user.has_perm(object_policy.codename, obj)):
-        if method != 'DELETE':
+    elif user.is_staff:
+        has_permission, message = True, None
+    else:
+        permission = app.app_permission_registry.get(method)
+        if permission and user.has_perm(permission.codename, obj):
             has_permission, message = True, None
-        else:
-            has_permission, message = False, "{user} does not have permission to delete."
-    elif has_permission and isinstance(obj, app.App):
+    if has_permission and isinstance(obj, app.App):
         return get_app_status(obj)
     return has_permission, message
 
@@ -82,7 +81,7 @@ class IsObjectUser(permissions.BasePermission):
     an app-related model.
     """
     def has_object_permission(self, request, view, obj):
-        return has_object_permission(request.user, obj, request.method)[0]
+        return has_app_permission(request.user, obj, request.method)[0]
 
 
 class IsAdmin(permissions.BasePermission):

rootfs/api/models/certificate.py

@@ -506,6 +506,7 @@ def validate_target_port(cls, value):
 class CertificateSerializer(serializers.ModelSerializer):
     """Serialize a :class:`~api.models.certificate.Certificate` model."""
 
+    app = serializers.SlugRelatedField(slug_field='id', queryset=models.app.App.objects.all())
     owner = serializers.ReadOnlyField(source='owner.username')
     domains = serializers.ReadOnlyField()
     san = serializers.ListField(