chore(perms): add object perms

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_app_permission
+from .permissions import has_object_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_app_permission(self.scope["user"], app, "GET")
+                permission = has_object_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

@@ -0,0 +1,17 @@
+# Generated by Django 4.2.15 on 2024-08-14 09:43
+
+from django.db import migrations
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('api', '0009_remove_appsettings_canaries_remove_release_canary_and_more'),
+    ]
+
+    operations = [
+        migrations.AlterModelOptions(
+            name='certificate',
+            options={'ordering': ['name', 'common_name', 'expires'], 'permissions': (('use_cert', 'Can use cert'),)},
+        ),
+    ]

rootfs/api/models/app.py

@@ -31,10 +31,12 @@
 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
+from .base import UuidAuditedModel, PROCFILE_TYPE_WEB, PROCFILE_TYPE_RUN, DEFAULT_HTTP_PORT, \
+    ObjectPolicy, object_policy_registry
 
 User = get_user_model()
 logger = logging.getLogger(__name__)
+app_policy = ObjectPolicy('id', 'use_app', 'Can use app')  # query field, policy, description
 
 
 # http://kubernetes.io/v1.1/docs/design/identifiers.html
@@ -77,7 +79,7 @@ class App(UuidAuditedModel):
 
     class Meta:
         verbose_name = 'Application'
-        permissions = (('use_app', 'Can use app'),)
+        permissions = (app_policy[1:], )
         ordering = ['id']
 
     def save(self, *args, **kwargs):
@@ -1161,3 +1163,7 @@ 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/models/base.py

@@ -4,6 +4,7 @@
 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
@@ -38,6 +39,38 @@
 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/certificate.py

@@ -15,11 +15,12 @@
 from api.utils import validate_label
 from api.exceptions import AlreadyExists, ServiceUnavailable
 from scheduler import KubeException
-from .base import AuditedModel
+from .base import AuditedModel, ObjectPolicy, object_policy_registry
 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.
@@ -120,6 +121,7 @@ class Certificate(AuditedModel):
     subject = models.TextField(editable=False)
 
     class Meta:
+        permissions = (cert_policy[1:], )
         ordering = ['name', 'common_name', 'expires']
 
     @property
@@ -238,3 +240,7 @@ 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/permissions.py

@@ -3,37 +3,38 @@
 from django.conf import settings
 from django.contrib.auth.models import AnonymousUser
 from api import manager
-from api.models.blocklist import Blocklist, App
+from api.models import app
+from api.models import blocklist
+from api.models.base import object_policy_registry
 
 
 def get_app_status(app):
-    blocklist = Blocklist.get_blocklist(app)
-    if blocklist:
-        return False, blocklist.remark
+    block = blocklist.Blocklist.get_blocklist(app)
+    if block:
+        return False, block.remark
     if settings.WORKFLOW_MANAGER_URL:
         status = manager.User().get_status(app.owner.pk)
         if not status["is_active"]:
             return False, status["message"]
     return True, None
 
 
-def has_app_permission(user, obj, method):
-    if isinstance(obj, App) or hasattr(obj, 'app'):
-        app = obj if isinstance(obj, App) else obj.app
-        is_ok, message = get_app_status(app)
-        if is_ok:
-            if user.is_superuser:
-                return True, None
-            elif app.owner == user:
-                return True, None
-            elif user.is_staff or user.has_perm('use_app', app):
-                if method != 'DELETE':
-                    return True, None
-                else:
-                    return False, "User does not have permission to delete"
+def has_object_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':
+            has_permission, message = True, None
         else:
-            return is_ok, message
-    return False, "App object does not exist or does not have permission."
+            has_permission, message = False, "{user} does not have permission to delete."
+    elif has_permission and isinstance(obj, app.App):
+        return get_app_status(obj)
+    return has_permission, message
 
 
 class IsAnonymous(permissions.BasePermission):
@@ -75,13 +76,13 @@ def has_object_permission(self, request, view, obj):
             return False
 
 
-class IsAppUser(permissions.BasePermission):
+class IsObjectUser(permissions.BasePermission):
     """
     Object-level permission to allow owners or collaborators to access
     an app-related model.
     """
     def has_object_permission(self, request, view, obj):
-        return has_app_permission(request.user, obj, request.method)[0]
+        return has_object_permission(request.user, obj, request.method)[0]
 
 
 class IsAdmin(permissions.BasePermission):

rootfs/api/tests/test_app.py

@@ -22,6 +22,7 @@
 
 from api.exceptions import DryccException
 from api.tests import adapter, DryccTestCase
+from api.models.base import object_policy_registry
 import requests_mock
 
 User = get_user_model()
@@ -358,9 +359,12 @@ def test_app_transfer(self, mock_requests):
         self.assertEqual(config2.owner, collaborator)
 
         # Collaborators can't transfer
-        body = {'username': owner.username}
-        perms_url = url+"/perms/"
-        response = self.client.post(perms_url, body)
+        body = {
+            'username': owner.username,
+            'codename': object_policy_registry.get(App)[1].codename,
+            'uniqueid': app.id,
+        }
+        response = self.client.post('/v2/perms/rules/', body)
         self.assertEqual(response.status_code, 201, response.data)
 
         self.client.credentials(HTTP_AUTHORIZATION='Token ' + owner_token)

rootfs/api/tests/test_hooks.py

@@ -8,6 +8,8 @@
 from django.core.cache import cache
 
 from api.tests import adapter, DryccTransactionTestCase
+from api.models.app import App
+from api.models.base import object_policy_registry
 import requests_mock
 
 User = get_user_model()
@@ -70,8 +72,12 @@ def test_key_hook(self, mock_requests):
         app_id = self.create_app()
 
         # give user permission to app
-        url = "/v2/apps/{}/perms".format(app_id)
-        body = {'username': str(self.user)}
+        body = {
+            'username': str(self.user),
+            'codename': object_policy_registry.get(App)[1].codename,
+            'uniqueid': app_id,
+        }
+        url = "/v2/perms/rules"
         response = self.client.post(url, body)
         self.assertEqual(response.status_code, 201, response.data)