Permissions attribute for targets, filtering logic and migrations

tom_targets/base_models.py

@@ -290,6 +290,11 @@ class BaseTarget(models.Model):
         ('JPL_MAJOR_PLANET', 'JPL Major Planet')
     )
 
+    class Permissions(models.TextChoices):
+        OPEN = 'OPEN'
+        PUBLIC = 'PUBLIC'
+        PRIVATE = 'PRIVATE'
+
     name = models.CharField(
         max_length=100, default='', verbose_name='Name', help_text='The name of this target e.g. Barnard\'s star.',
         unique=True
@@ -306,6 +311,10 @@ class BaseTarget(models.Model):
         auto_now=True, verbose_name='Last Modified',
         help_text='The time which this target was changed in the TOM database.'
     )
+    permissions = models.CharField(
+        max_length=100, default=Permissions.PUBLIC, choices=Permissions.choices,
+        help_text='The access level of this target, see the docs on public vs private targets.'
+    )
     ra = models.FloatField(
         null=True, blank=True, verbose_name='Right Ascension', help_text='Right Ascension, in degrees.'
     )

tom_targets/migrations/0024_basetarget_permissions.py

@@ -0,0 +1,18 @@
+# Generated by Django 4.2.18 on 2025-02-06 18:39
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('tom_targets', '0023_alter_basetarget_created'),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='basetarget',
+            name='permissions',
+            field=models.CharField(choices=[('OPEN', 'Open'), ('PUBLIC', 'Public'), ('PRIVATE', 'Private')], default='PUBLIC', help_text='The acess level of this target, see the docs on public vs private targets.', max_length=100),
+        ),
+    ]

tom_targets/migrations/0025_auto_20250206_2017.py

@@ -0,0 +1,58 @@
+# Generated by Django 4.2.18 on 2025-02-06 20:17
+
+from django.db import migrations
+from django.conf import settings
+
+def get_target_model():
+    try:
+        custom_class = settings.TARGET_MODEL_CLASS
+        return custom_class.split('.')[0], custom_class.split('.')[-1]
+    except AttributeError:
+        return 'tom_targets', 'BaseTarget'
+
+def remove_public_group(apps, schema_editor):
+    target_app, target_model = get_target_model()
+    Group = apps.get_model('auth', 'Group')
+    Target = apps.get_model(target_app, target_model)
+    UserObjectPermission = apps.get_model('guardian', 'UserObjectPermission')
+    GroupObjectPermission = apps.get_model('guardian', 'GroupObjectPermission')
+
+    group, _ = Group.objects.get_or_create(name='Public')
+
+    # Delete Target permissions for public group
+    GroupObjectPermission.objects.filter(group=group, content_type__model=target_model.lower()).delete()
+
+    # Any remaining permissions means target should be private
+    private_group_permissions = GroupObjectPermission.objects.filter(
+        content_type__model=target_model.lower()
+    )
+    private_user_permissions = UserObjectPermission.objects.filter(
+        content_type__model=target_model.lower()
+    )
+
+    # get a list of target ids that still have permissions
+    target_ids = set(
+        list(private_group_permissions.values_list('object_pk', flat=True)) \
+        + list(private_user_permissions.values_list('object_pk', flat=True))
+    )
+
+    # Update targets to private
+    Target.objects.filter(pk__in=target_ids).update(permissions='PRIVATE')
+
+    # Delete public group
+    group.delete()
+
+def set_all_to_public(apps, schema_editor):
+    target_app, target_model = get_target_model()
+    Target = apps.get_model(target_app, target_model)
+    Target.objects.update(permissions='PUBLIC')
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('tom_targets', '0024_basetarget_permissions'),
+    ]
+
+    operations = [
+        migrations.RunPython(remove_public_group, set_all_to_public),
+    ]

tom_targets/tests/tests.py

@@ -2,7 +2,7 @@
 from datetime import datetime
 import responses
 
-from django.contrib.auth.models import User, Group
+from django.contrib.auth.models import AnonymousUser, User, Group
 from django.contrib.messages import get_messages
 from django.contrib.messages.constants import SUCCESS, WARNING
 from django.core.files.uploadedfile import SimpleUploadedFile
@@ -18,6 +18,7 @@
 from tom_targets.models import Target, TargetExtra, TargetList, TargetName
 from tom_targets.utils import import_targets
 from tom_targets.merge import target_merge
+from tom_targets.views import target_permission_filter
 from tom_dataproducts.models import ReducedDatum, DataProduct
 from tom_observations.models import ObservationRecord
 from guardian.shortcuts import assign_perm, get_perms
@@ -1946,3 +1947,66 @@ def test_seed_targets_unauthenticated(self):
         response = self.client.post(reverse('targets:seed'))
         self.assertEqual(response.status_code, 302)
         self.assertFalse(Target.objects.exists())
+
+
+class TestTargetPermissionFiltering(TestCase):
+    def setUp(self):
+        self.user = User.objects.create(username='testuser')
+        self.group = Group.objects.create(name='testgroup')
+        self.open_target = SiderealTargetFactory.create(permissions=Target.Permissions.OPEN)
+        self.public_target = SiderealTargetFactory.create(permissions=Target.Permissions.PUBLIC)
+        self.private_group_target = SiderealTargetFactory.create(permissions=Target.Permissions.PRIVATE)
+        self.private_user_target = SiderealTargetFactory.create(permissions=Target.Permissions.PRIVATE)
+
+    def test_open_targets_visible(self):
+        result = target_permission_filter(AnonymousUser(), Target.objects.all(), 'view_target')
+        self.assertIn(self.open_target, result)
+        self.assertNotIn(self.public_target, result)
+        self.assertNotIn(self.private_group_target, result)
+        self.assertNotIn(self.private_user_target, result)
+
+    def test_public_targets_visible(self):
+        result = target_permission_filter(self.user, Target.objects.all(), 'view_target')
+        self.assertIn(self.open_target, result)
+        self.assertIn(self.public_target, result)
+        self.assertNotIn(self.private_group_target, result)
+        self.assertNotIn(self.private_user_target, result)
+
+    def test_private_group_permission(self):
+        self.group.user_set.add(self.user)
+        assign_perm('tom_targets.view_target', self.group, self.private_group_target)
+        result = target_permission_filter(self.user, Target.objects.all(), 'view_target')
+        self.assertIn(self.open_target, result)
+        self.assertIn(self.public_target, result)
+        self.assertIn(self.private_group_target, result)
+        self.assertNotIn(self.private_user_target, result)
+
+    def test_private_user_permission(self):
+        assign_perm('tom_targets.view_target', self.user, self.private_user_target)
+        result = target_permission_filter(self.user, Target.objects.all(), 'view_target')
+        self.assertIn(self.open_target, result)
+        self.assertIn(self.public_target, result)
+        self.assertNotIn(self.private_group_target, result)
+        self.assertIn(self.private_user_target, result)
+
+    def test_private_user_permission_wrong_action(self):
+        assign_perm('tom_targets.view_target', self.user, self.private_user_target)
+        result = target_permission_filter(self.user, Target.objects.all(), 'delete_target')
+        self.assertIn(self.open_target, result)
+        self.assertIn(self.public_target, result)
+        self.assertNotIn(self.private_group_target, result)
+        self.assertNotIn(self.private_user_target, result)
+
+    def test_superuser_permission(self):
+        self.user.is_superuser = True
+        self.user.save()
+        result = target_permission_filter(self.user, Target.objects.all(), 'view_target')
+        self.assertIn(self.open_target, result)
+        self.assertIn(self.public_target, result)
+        self.assertIn(self.private_group_target, result)
+        self.assertIn(self.private_user_target, result)
+
+    def test_typo_in_action(self):
+        """Make sure a typo doesn't expose data"""
+        with self.assertRaises(AssertionError):
+            target_permission_filter(self.user, Target.objects.all(), 'view_targett')

tom_targets/views.py

@@ -61,8 +61,23 @@
 logger = logging.getLogger(__name__)
 logger.setLevel(logging.DEBUG)
 
+def target_permission_filter(user, qs, action):
+    assert action in ['view_target', 'change_target', 'delete_target']
 
-class TargetListView(PermissionListMixin, FilterView):
+    if user.is_authenticated:
+        if user.is_superuser:
+            # Do not filter the queryset by permissions at all
+            return qs
+        else:
+            # Exclude targets that are private except for those that the user has explicit permissions to view
+            private_targets = qs.filter(permissions=Target.Permissions.PRIVATE)
+            public_targets = qs.exclude(permissions=Target.Permissions.PRIVATE)
+            return public_targets | get_objects_for_user(user, f'{Target._meta.app_label}.{action}', private_targets)
+    else:
+        # Only allow open targets
+        return qs.exclude(permissions__in=[Target.Permissions.PUBLIC, Target.Permissions.PRIVATE])
+
+class TargetListView(FilterView):
     """
     View for listing targets in the TOM. Only shows targets that the user is authorized to view. Requires authorization.
     """
@@ -72,7 +87,6 @@ class TargetListView(PermissionListMixin, FilterView):
     model = Target
     filterset_class = TargetFilter
     # Set app_name for Django-Guardian Permissions in case of Custom Target Model
-    permission_required = f'{Target._meta.app_label}.view_target'
     ordering = ['-created']
 
     def get_context_data(self, *args, **kwargs):
@@ -93,6 +107,10 @@ def get_context_data(self, *args, **kwargs):
         context['query_string'] = self.request.META['QUERY_STRING']
         return context
 
+    def get_queryset(self, *args, **kwargs):
+        qs = super().get_queryset(*args, **kwargs)
+        return target_permission_filter(self.request.user, qs, 'view_target')
+
 
 class TargetNameSearchView(RedirectView):
     """