Synchronize the created and add object_created field

Author: AlanCoding

ansible_base/rbac/migrations/0004_remote_permissions_additions.py

@@ -3,6 +3,7 @@
 import ansible_base.rbac.models.content_type
 import ansible_base.rbac.remote
 from django.db import migrations, models
+import django.utils.timezone
 
 
 class Migration(migrations.Migration):
@@ -159,6 +160,28 @@ class Migration(migrations.Migration):
             model_name='roleevaluationuuid',
             name='dab_rbac_ro_role_id_4fe905_idx',
         ),
+        # Added field for a checksum like purpose for remote models
+        migrations.AddField(
+            model_name='roleuserassignment',
+            name='object_created',
+            field=models.DateTimeField(help_text='The created timestamp of related object, if applicable.', null=True),
+        ),
+        migrations.AddField(
+            model_name='roleteamassignment',
+            name='object_created',
+            field=models.DateTimeField(help_text='The created timestamp of related object, if applicable.', null=True),
+        ),
+        # Make assignment created timestamp backdateable
+        migrations.AlterField(
+            model_name='roleteamassignment',
+            name='created',
+            field=models.DateTimeField(default=django.utils.timezone.now, editable=False, help_text='The date/time this resource was created.'),
+        ),
+        migrations.AlterField(
+            model_name='roleuserassignment',
+            name='created',
+            field=models.DateTimeField(default=django.utils.timezone.now, editable=False, help_text='The date/time this resource was created.'),
+        ),
         # Fields unique to DAB RBAC and not generally shared with ContentType or Permission
         migrations.AddField(
             model_name='dabcontenttype',

ansible_base/rbac/models/role.py

@@ -1,5 +1,6 @@
 import logging
 from collections.abc import Iterable
+from datetime import datetime
 from typing import Optional, Type, Union
 from uuid import UUID
 
@@ -9,6 +10,7 @@
 from django.db.models.functions import Cast
 from django.db.models.query import QuerySet
 from django.db.utils import IntegrityError
+from django.utils import timezone
 from django.utils.translation import gettext_lazy as _
 
 # Django-rest-framework
@@ -63,6 +65,18 @@ def __getattr__(self, attr):
             return rd
 
 
+def get_created_timestamp(obj: Union[models.Model, RemoteObject]) -> Optional[datetime]:
+    """Given some obj from the users app, try to infer the created timestamp"""
+    if isinstance(obj, RemoteObject):
+        return obj.created
+    for field_name in ('created', 'created_at'):
+        if hasattr(obj, field_name):
+            val = getattr(obj, field_name)
+            if isinstance(val, datetime):
+                return val
+    return None
+
+
 class RoleDefinitionManager(models.Manager):
     def contribute_to_class(self, cls: Type[models.Model], name: str) -> None:
         """After Django populates the model for the manager, attach the manager role manager"""
@@ -178,13 +192,13 @@ def __str__(self):
             managed_str = ', managed=True'
         return f'RoleDefinition(pk={self.id}, name={self.name}{managed_str})'
 
-    def give_global_permission(self, actor):
-        return self.give_or_remove_global_permission(actor, giving=True)
+    def give_global_permission(self, actor, assignment_created=None):
+        return self.give_or_remove_global_permission(actor, giving=True, assignment_created=assignment_created)
 
     def remove_global_permission(self, actor):
         return self.give_or_remove_global_permission(actor, giving=False)
 
-    def give_or_remove_global_permission(self, actor, giving=True):
+    def give_or_remove_global_permission(self, actor, giving=True, assignment_created=None):
         if giving and (self.content_type is not None):
             raise ValidationError('Role definition content type must be null to assign globally')
 
@@ -202,6 +216,8 @@ def give_or_remove_global_permission(self, actor, giving=True):
             raise RuntimeError(f'Cannot {giving and "give" or "remove"} permission for {actor}, must be a user or team')
 
         if giving:
+            if assignment_created:
+                kwargs['created'] = assignment_created
             assignment, _ = cls.objects.get_or_create(**kwargs)
         else:
             assignment = cls.objects.filter(**kwargs).first()
@@ -221,8 +237,8 @@ def give_or_remove_global_permission(self, actor, giving=True):
 
         return assignment
 
-    def give_permission(self, actor, content_object):
-        return self.give_or_remove_permission(actor, content_object, giving=True)
+    def give_permission(self, actor, content_object, assignment_created=None):
+        return self.give_or_remove_permission(actor, content_object, giving=True, assignment_created=assignment_created)
 
     def remove_permission(self, actor, content_object):
         return self.give_or_remove_permission(actor, content_object, giving=False)
@@ -247,7 +263,7 @@ def get_or_create_object_role(self, kwargs, defaults):
             object_role = ObjectRole.objects.create(**kwargs, **defaults)
             return (object_role, True)
 
-    def give_or_remove_permission(self, actor, content_object, giving=True, sync_action=False):
+    def give_or_remove_permission(self, actor, content_object, giving=True, sync_action=False, assignment_created=None):
         "Shortcut method to do whatever needed to give user or team these permissions"
         validate_assignment(self, actor, content_object)
 
@@ -278,15 +294,28 @@ def give_or_remove_permission(self, actor, content_object, giving=True, sync_act
 
         update_teams, to_update = needed_updates_on_assignment(self, actor, object_role, created=created, giving=True)
 
+        assignment_defaults = {}
+        object_created = get_created_timestamp(content_object)
+        if object_created:
+            assignment_defaults['object_created'] = object_created
+        if assignment_created:
+            assignment_defaults['created'] = assignment_created
+
         assignment = None
         if actor._meta.model_name == 'user':
             if giving:
-                assignment, created = RoleUserAssignment.objects.get_or_create(user=actor, object_role=object_role)
+                try:
+                    assignment = RoleUserAssignment.objects.get(user=actor, object_role=object_role)
+                except RoleUserAssignment.DoesNotExist:
+                    assignment = RoleUserAssignment.objects.create(user=actor, object_role=object_role, **assignment_defaults)
             else:
                 object_role.users.remove(actor)
         elif isinstance(actor, permission_registry.team_model):
             if giving:
-                assignment, created = RoleTeamAssignment.objects.get_or_create(team=actor, object_role=object_role)
+                try:
+                    assignment = RoleTeamAssignment.objects.get(team=actor, object_role=object_role)
+                except RoleTeamAssignment.DoesNotExist:
+                    assignment = RoleTeamAssignment.objects.create(team=actor, object_role=object_role, **assignment_defaults)
             else:
                 object_role.teams.remove(actor)
 
@@ -410,6 +439,14 @@ class AssignmentBase(ImmutableCommonModel, ObjectRoleFields):
         null=True, blank=True, help_text=_('The primary key of the object this assignment applies to; null value indicates system-wide assignment.')
     )
     content_type = models.ForeignKey(DABContentType, on_delete=models.CASCADE, null=True, help_text=_("The content type this applies to."))
+    # The object_created field can be used for a checksum-like purpose to verify nothing strange happened with the related object
+    object_created = models.DateTimeField(help_text=_("The created timestamp of related object, if applicable."), null=True)
+    # Define this with default to make it possible to backdate if necessary, for sync
+    created = models.DateTimeField(
+        default=timezone.now,
+        editable=False,
+        help_text=_("The date/time this resource was created."),
+    )
 
     # object_role is internal, and not shown in serializer
     # content_type does not have a link, and ResourceType will be used in lieu sometime

ansible_base/rbac/remote.py

@@ -58,9 +58,11 @@ def __init__(self, ct: models.Model, abstract=False):
 class RemoteObject:
     """Placeholder for objects that live in another project."""
 
-    def __init__(self, content_type: models.Model, object_id: Union[int, str], parent_reference=None):
+    def __init__(self, content_type: models.Model, object_id: Union[int, str], parent_reference=None, created=None):
         self.content_type = content_type
         self.object_id = object_id
+        # Allow tracking details of the object
+        self.created = created
         # Since object is remote, we do not have its properties here, so a pointer to the parent can be specified here
         self.parent_reference = parent_reference
         if not hasattr(self, '_meta'):

ansible_base/rbac/service_api/serializers.py

@@ -57,6 +57,8 @@ class BaseAssignmentSerializer(serializers.ModelSerializer):
     object_ansible_id = ObjectIDAnsibleIDField(source='object_id', required=False, allow_null=True)
     object_id = serializers.CharField(allow_blank=True, required=False, allow_null=True)
     from_service = serializers.CharField(write_only=True)
+    # Force created field to be writable
+    created = serializers.DateTimeField(required=False)
 
     def to_representation(self, instance):
         # hack to surface content_object for ObjectIDAnsibleIDField
@@ -131,10 +133,10 @@ def create(self, validated_data):
                     raise serializers.ValidationError({'object_id': _('Object must be specified for this role assignment')})
 
                 with transaction.atomic():
-                    assignment = rd.give_permission(actor, obj)
+                    assignment = rd.give_permission(actor, obj, assignment_created=validated_data.get('created'))
             else:
                 with transaction.atomic():
-                    assignment = rd.give_global_permission(actor)
+                    assignment = rd.give_global_permission(actor, assignment_created=validated_data.get('created'))
 
             return assignment
 

test_app/tests/rbac/remote/test_remote_assignment.py

@@ -143,3 +143,61 @@ def test_org_roles_same_type_different_service(rando, organization):
         ], f'User should have permission to exactly {service_name} resource'
 
         rds[service_name].remove_permission(rando, organization)
+
+
+@pytest.mark.django_db
+def test_object_created_field_local_model(rando, org_inv_rd, organization):
+    """
+    Test that the object_created field is set to the local object's created timestamp
+    when creating an assignment for a local model like inventory.
+
+    This test should FAIL because the object_created field doesn't exist yet.
+    """
+    # Create the assignment
+    assignment = org_inv_rd.give_permission(rando, organization)
+
+    # Verify the assignment was created
+    assert assignment.user == rando
+    assert assignment.role_definition == org_inv_rd
+    assert assignment.object_id == organization.pk
+
+    # This should FAIL - the object_created field should be set to inventory.created
+    assert hasattr(assignment, 'object_created'), "Assignment should have an object_created field"
+
+    assert assignment.object_created == organization.created, (
+        f"object_created should be set to the organization's created timestamp. "
+        f"Expected: {organization.created}, but object_created field is missing or has wrong value"
+    )
+
+
+@pytest.mark.django_db
+def test_object_created_field_remote_object(rando, foo_type, foo_rd):
+    """
+    Test that the object_created field is properly handled when creating an assignment
+    for a remote object (stand-in object pattern).
+
+    This test should FAIL because the object_created field doesn't exist yet.
+    """
+    from datetime import datetime, timezone
+
+    # Create a remote object stand-in
+    remote_foo = RemoteObject(content_type=foo_type, object_id=42)
+
+    # Create the assignment
+    assignment = foo_rd.give_permission(rando, remote_foo)
+
+    # Verify the assignment was created
+    assert assignment.user == rando
+    assert assignment.role_definition == foo_rd
+    assert assignment.object_id == 42
+    assert isinstance(assignment.content_object, RemoteObject)
+
+    # This should FAIL - the object_created field should exist and be handled for remote objects
+    assert hasattr(assignment, 'object_created'), "Assignment should have an object_created field even for remote objects"
+
+    # For remote objects, the object_created field might be None or set to a default value
+    # since we don't have the actual remote object's creation timestamp
+    # The exact behavior will depend on implementation, but the field should exist
+    assert assignment.object_created is not None or assignment.object_created is None, (
+        f"object_created field should exist for remote objects. " f"Current value: {getattr(assignment, 'object_created', 'FIELD_MISSING')}"
+    )

test_app/tests/rbac/remote/test_service_api.py

@@ -3,7 +3,7 @@
 import pytest
 
 from ansible_base.lib.utils.response import get_relative_url
-from ansible_base.rbac.models import DABContentType, DABPermission, RoleDefinition
+from ansible_base.rbac.models import DABContentType, DABPermission, RoleDefinition, RoleTeamAssignment, RoleUserAssignment
 from test_app.models import Team, User
 
 
@@ -287,3 +287,53 @@ def test_service_api_permissions(reverse_name, normal_case, unauth_case, admin_a
 
     unauth_response = unauthenticated_api_client.get(url)
     assert unauth_response.status_code == unauth_case, unauth_response.data
+
+
+@pytest.mark.django_db
+def test_service_assignment_created_timestamp_sync(admin_api_client, rando, inv_rd, inventory):
+    """
+    Test that demonstrates the field sync issue: the 'created' timestamp field is displayed
+    in responses but not applied when creating assignments via POST to /assign/.
+
+    This test should FAIL, showing that custom timestamps are ignored and auto-generated instead.
+    """
+    from datetime import datetime, timezone
+
+    from django.utils.dateparse import parse_datetime
+
+    url = get_relative_url('serviceuserassignment-assign')
+
+    creator_user = User.objects.create(username='timestamp_creator')
+
+    # Set a specific timestamp that's different from "now"
+    custom_timestamp = datetime(2023, 1, 15, 10, 30, 45, tzinfo=timezone.utc)
+    custom_timestamp_str = custom_timestamp.isoformat()
+
+    post_data = {
+        "role_definition": inv_rd.name,
+        "user_ansible_id": str(rando.resource.ansible_id),
+        "object_id": str(inventory.pk),
+        "created_by_ansible_id": str(creator_user.resource.ansible_id),
+        "created": custom_timestamp_str,
+        "from_service": "test_service",
+    }
+
+    response = admin_api_client.post(url, data=post_data)
+    assert response.status_code == 201, response.data
+
+    assignment = RoleUserAssignment.objects.get(user=rando, role_definition=inv_rd, object_id=inventory.pk)
+
+    # Test if the custom timestamp was properly set
+    expected_created = custom_timestamp
+    actual_created = assignment.created
+
+    # This should FAIL, demonstrating the field sync issue
+    assert actual_created == expected_created, (
+        f"FIELD SYNC ISSUE: Expected created timestamp '{expected_created}' but got '{actual_created}'. "
+        f"The 'created' field is displayed in responses but not applied from POST data."
+    )
+
+    # Verify response contains the timestamp field (showing it's "displayed")
+    response_created = parse_datetime(response.data['created'])
+    # Note: This will show the auto-generated timestamp, not our custom one
+    assert response_created == expected_created, f"Response created timestamp should match: expected '{expected_created}' but got '{response_created}'"