Merge branch 'ansible:devel' into aap-52229-activity-stream

Author: arrestle

.github/workflows/ci.yml

@@ -18,8 +18,8 @@ jobs:
             python-version: "3.11"
             sonar: false
             junit-xml-upload: false
-          - env: py39
-            python-version: "3.9"
+          - env: py312
+            python-version: "3.12"
             sonar: false
             junit-xml-upload: false
           - env: py310

.github/workflows/linting.yml

@@ -22,6 +22,8 @@ jobs:
             command: check_black
           - name: api-isort
             command: check_isort
+          - name: pure-python-imports
+            command: check_pure_python_imports
     steps:
       - name: Install make
         run: sudo apt install make
@@ -38,5 +40,18 @@ jobs:
       - name: Install requirments
         run: pip3.11 install -r requirements/requirements_dev.txt
 
+      - name: Install test dependencies for pure-python-imports
+        if: matrix.tests.name == 'pure-python-imports'
+        run: |
+          pip3.11 install -r requirements/requirements.in
+          pip3.11 install -r requirements/requirements_testing.txt
+          pip3.11 install -r requirements/requirements_channels.in
+          pip3.11 install -r requirements/requirements_redis_client.in
+
+      - name: Run pure-python-imports check
+        if: matrix.tests.name == 'pure-python-imports'
+        run: python tools/scripts/check_pure_python_imports.py
+
       - name: Run check ${{ matrix.tests.name }}
+        if: matrix.tests.name != 'pure-python-imports'
         run: make ${{ matrix.tests.command }}

Makefile

@@ -49,6 +49,7 @@ check_flake8:
 check_isort:
 	tox -e isort -- --check $(CHECK_SYNTAX_FILES)
 
+
 ## Starts a postgres container in the background if one is not running
 # Options:
 #  -d, --detatch: run the container in background

ansible_base/jwt_consumer/common/util.py

@@ -7,7 +7,6 @@
 from cryptography.hazmat.primitives import hashes, serialization
 from cryptography.hazmat.primitives.asymmetric import padding
 
-from ansible_base.jwt_consumer.common.cert import JWTCert, JWTCertException
 from ansible_base.lib.utils.settings import get_setting
 
 logger = logging.getLogger('ansible_base.jwt_consumer.common.util')
@@ -32,6 +31,8 @@ def generate_x_trusted_proxy_header(key: str) -> str:
 
 
 def validate_x_trusted_proxy_header(header_value: str, ignore_cache=False) -> bool:
+    from ansible_base.jwt_consumer.common.cert import JWTCert, JWTCertException
+
     try:
         cert = JWTCert()
         cert.get_decryption_key(ignore_cache=ignore_cache)

ansible_base/lib/utils/create_system_user.py

@@ -1,20 +1,22 @@
-import logging
-from typing import Optional, Tuple, Type
+from __future__ import annotations
 
-from django.core.exceptions import ImproperlyConfigured
-from django.db import models
-from django.utils.translation import gettext as _
+import logging
+from typing import TYPE_CHECKING, Optional, Tuple, Type
 
 from ansible_base.lib.utils.settings import get_setting
 
+if TYPE_CHECKING:
+    from django.db import models
+
 logger = logging.getLogger('ansible_base.lib.utils.create_system_user')
 
 """
 These functions are in its own file because it is loaded during migrations so it has no access to models.
 """
 
 
-def create_system_user(user_model: Type[models.Model]) -> models.Model:  # Note: We can't load models here so we can typecast to anything better than Model
+def create_system_user(user_model: Type[models.Model]) -> models.Model:
+    # Note: We can't load models here so we can typecast to anything better than Model
     from ansible_base.lib.abstract_models.user import AbstractDABUser
 
     #
@@ -63,4 +65,8 @@ def get_system_username() -> Tuple[Optional[str], str]:
         return str(value), setting_name
 
     logger.error(f"Expected get_setting to return a string for {setting_name}, got {type(value)}")
+
+    from django.core.exceptions import ImproperlyConfigured
+    from django.utils.translation import gettext as _
+
     raise ImproperlyConfigured(_("Setting %(setting_name)s needs to be a string not a %(type)s") % {'setting_name': setting_name, 'type': type(value)})

ansible_base/lib/utils/requests.py

@@ -1,12 +1,13 @@
-import logging
-from typing import Optional
+from __future__ import annotations
 
-from crum import get_current_request
-from django.http import HttpRequest
+import logging
+from typing import TYPE_CHECKING, Optional
 
-from ansible_base.jwt_consumer.common.util import validate_x_trusted_proxy_header
 from ansible_base.lib.utils.settings import get_setting
 
+if TYPE_CHECKING:
+    from django.http import HttpRequest
+
 logger = logging.getLogger('ansible_base.lib.uitls.requests')
 
 
@@ -43,6 +44,8 @@ def get_remote_hosts(request: HttpRequest, get_first_only: bool = False) -> list
     # If we are connected to from a trusted proxy then we can add some additional headers
     try:
         if 'HTTP_X_TRUSTED_PROXY' in request.META:
+            from ansible_base.jwt_consumer.common.util import validate_x_trusted_proxy_header
+
             if validate_x_trusted_proxy_header(request.META['HTTP_X_TRUSTED_PROXY']):
                 # The last entry in x-forwarded-for from envoy can be trusted implicitly
                 # https://www.envoyproxy.io/docs/envoy/latest/configuration/http/http_conn_man/headers#x-forwarded-for
@@ -68,10 +71,14 @@ def get_remote_hosts(request: HttpRequest, get_first_only: bool = False) -> list
 def is_proxied_request(request: Optional[HttpRequest] = None) -> bool:
     "Return true if request claims to be from a proxy and the header validates as such."
     if request is None:
+        from crum import get_current_request
+
         request = get_current_request()
     if request is None:
         # e.g. being called by CLI or something
         return False
     if x_trusted_proxy := request.META.get("HTTP_X_TRUSTED_PROXY"):
+        from ansible_base.jwt_consumer.common.util import validate_x_trusted_proxy_header
+
         return validate_x_trusted_proxy_header(x_trusted_proxy)
     return False

ansible_base/rbac/migrations/0005_remote_permissions_data.py

@@ -1,62 +1,8 @@
 # Generated by Django 4.2.23 on 2025-06-30 12:48
 
-import logging
-
 from django.db import migrations
 
-
-logger = logging.getLogger(__name__)
-
-
-def create_types_if_needed(apps, schema_editor):
-    """Before we can migrate to the new DABContentType, entries in that table must be created.
-
-    This method runs what is ordinarily the post_migrate logic, but in the migration case here.
-    Only needed in the upgrade case, otherwise better to run at true post-migrate.
-    """
-    permission_cls = apps.get_model('dab_rbac', 'DABPermission')
-    rd_cls = apps.get_model('dab_rbac', 'RoleDefinition')
-    if permission_cls.objects.exists() or rd_cls.objects.exists():
-        logger.info('Running DABContentType creation script as part of 0005 migration')
-        from ansible_base.rbac.management.create_types import create_DAB_contenttypes
-
-        create_DAB_contenttypes(apps=apps)
-
-
-def migrate_content_type(apps, schema_editor):
-    ct_cls = apps.get_model('dab_rbac', 'DABContentType')
-    ct_cls.objects.clear_cache()
-    for model_name in ('dabpermission', 'objectrole', 'roledefinition', 'roleuserassignment', 'roleteamassignment'):
-        cls = apps.get_model('dab_rbac', model_name)
-        update_ct = 0
-        for obj in cls.objects.all():
-            old_ct = obj.content_type
-            if old_ct:
-                try:
-                    # NOTE: could give duplicate normally, but that is impossible in migration path
-                    obj.new_content_type = ct_cls.objects.get_by_natural_key(old_ct.app_label, old_ct.model)
-                except Exception as e:
-                    raise RuntimeError(
-                        f"Failed to get new content type for a {model_name} pk={obj.pk}, obj={obj.__dict__}"
-                    ) from e
-                obj.save()
-                update_ct += 1
-        if update_ct:
-            logger.info(f'Updated content_type reference to new model for {model_name} for {update_ct} entries')
-    for model_name in ('roleevaluation', 'roleevaluationuuid'):
-        cls = apps.get_model('dab_rbac', model_name)
-        cls.objects.all().delete()
-
-    # DABPermission model had api_slug added in last migration
-    # if records existed before this point, it needs to be filled in
-    mod_ct = 0
-    permission_cls = apps.get_model('dab_rbac', 'DABPermission')
-    for permission in permission_cls.objects.all():
-        permission.api_slug = f'{permission.new_content_type.service}.{permission.codename}'
-        permission.save()
-        mod_ct += 1
-    if mod_ct:
-        logger.info(f'Set new field DABPermission.api_slug for {mod_ct} existing permissions')
+from . import _utils
 
 
 class Migration(migrations.Migration):
@@ -66,6 +12,6 @@ class Migration(migrations.Migration):
     ]
 
     operations = [
-        migrations.RunPython(create_types_if_needed, migrations.RunPython.noop),
-        migrations.RunPython(migrate_content_type, migrations.RunPython.noop),
+        migrations.RunPython(_utils.create_types_if_needed, migrations.RunPython.noop),
+        migrations.RunPython(_utils.migrate_content_type, migrations.RunPython.noop),
     ]

ansible_base/rbac/migrations/_utils.py

@@ -1,5 +1,10 @@
+# Generated by Claude Sonnet 4 (claude-sonnet-4@20250514)
+import logging
+
 from django.db import models
 
+logger = logging.getLogger(__name__)
+
 # This method has moved, and this is put here temporarily to make branch management easier
 from ansible_base.rbac.management import create_dab_permissions as create_custom_permissions  # noqa
 
@@ -45,3 +50,127 @@ def give_permissions(apps, rd, users=(), teams=(), object_id=None, content_type_
                 for team_id in teams
             ]
         RoleTeamAssignment.objects.bulk_create(team_assignments, ignore_conflicts=True)
+
+
+def cleanup_orphaned_permissions(apps):
+    """
+    Delete orphaned DABPermission objects for models no longer in the permission registry.
+
+    This is used during migrations to clean up permissions for any previously-registered model
+    that are no longer tracked by RBAC, but only if they are not referenced by any RoleDefinition.
+
+    Args:
+        apps: Django apps registry (from migration context)
+
+    Returns:
+        int: Number of permissions deleted
+    """
+    # Get model classes from apps registry
+    permission_cls = apps.get_model('dab_rbac', 'DABPermission')
+    role_definition_cls = apps.get_model('dab_rbac', 'RoleDefinition')
+
+    # Get permission registry to check which models are registered
+    from ansible_base.rbac import permission_registry
+    registered_model_keys = set()
+    for model in permission_registry._registry:
+        registered_model_keys.add((model._meta.app_label, model._meta.model_name))
+
+    # Find orphaned permissions
+    orphaned_permissions = []
+    for permission in permission_cls.objects.all():
+        if permission.content_type:
+            model_key = (permission.content_type.app_label, permission.content_type.model)
+            if model_key not in registered_model_keys:
+                # Check if this permission is referenced by any RoleDefinition
+                referencing_roles = role_definition_cls.objects.filter(permissions=permission)
+                if referencing_roles.exists():
+                    # Log warning for unregistered model still referenced by role definitions
+                    role_names = list(referencing_roles.values_list('name', flat=True))
+                    logger.warning(
+                        f'Permission {permission.codename} for unregistered model '
+                        f'{permission.content_type.app_label}.{permission.content_type.model} '
+                        f'is still referenced by role definitions: {role_names}'
+                    )
+                else:
+                    logger.info(f'Deleting orphaned permission {permission.codename} for unregistered model {model_key}')
+                    orphaned_permissions.append(permission)
+
+    # Delete orphaned permissions
+    deleted_count = 0
+    if orphaned_permissions:
+        deleted_count = len(orphaned_permissions)
+        permission_cls.objects.filter(id__in=[p.id for p in orphaned_permissions]).delete()
+        logger.info(f'Deleted {deleted_count} orphaned DABPermission objects for unregistered models')
+
+    return deleted_count
+
+
+def migrate_content_type(apps, schema_editor):
+    """
+    Migrate content type references from Django ContentType to DABContentType.
+
+    This function handles the migration of content type references across all RBAC models
+    from the old Django ContentType to the new DABContentType system.
+
+    Args:
+        apps: Django apps registry (from migration context)
+        schema_editor: Django schema editor (unused but required for migration signature)
+    """
+    # Pre-check: Delete orphaned DABPermission objects before migration
+    cleanup_orphaned_permissions(apps)
+
+    ct_cls = apps.get_model('dab_rbac', 'DABContentType')
+    ct_cls.objects.clear_cache()
+
+    for model_name in ('dabpermission', 'objectrole', 'roledefinition', 'roleuserassignment', 'roleteamassignment'):
+        cls = apps.get_model('dab_rbac', model_name)
+        update_ct = 0
+        for obj in cls.objects.all():
+            old_ct = obj.content_type
+            if old_ct:
+                try:
+                    # NOTE: could give duplicate normally, but that is impossible in migration path
+                    obj.new_content_type = ct_cls.objects.get_by_natural_key(old_ct.app_label, old_ct.model)
+                except Exception as e:
+                    raise RuntimeError(
+                        f"Failed to get new content type for a {model_name} pk={obj.pk}, obj={obj.__dict__}"
+                    ) from e
+                obj.save()
+                update_ct += 1
+        if update_ct:
+            logger.info(f'Updated content_type reference to new model for {model_name} for {update_ct} entries')
+    for model_name in ('roleevaluation', 'roleevaluationuuid'):
+        cls = apps.get_model('dab_rbac', model_name)
+        cls.objects.all().delete()
+
+    # DABPermission model had api_slug added in last migration
+    # if records existed before this point, it needs to be filled in
+    mod_ct = 0
+    permission_cls = apps.get_model('dab_rbac', 'DABPermission')
+    for permission in permission_cls.objects.all():
+        permission.api_slug = f'{permission.new_content_type.service}.{permission.codename}'
+        permission.save()
+        mod_ct += 1
+    if mod_ct:
+        logger.info(f'Set new field DABPermission.api_slug for {mod_ct} existing permissions')
+
+
+def create_types_if_needed(apps, schema_editor):
+    """
+    Create DABContentType entries if needed before migration.
+
+    Before we can migrate to the new DABContentType, entries in that table must be created.
+    This method runs what is ordinarily the post_migrate logic, but in the migration case here.
+    Only needed in the upgrade case, otherwise better to run at true post-migrate.
+
+    Args:
+        apps: Django apps registry (from migration context)
+        schema_editor: Django schema editor (unused but required for migration signature)
+    """
+    permission_cls = apps.get_model('dab_rbac', 'DABPermission')
+    rd_cls = apps.get_model('dab_rbac', 'RoleDefinition')
+    if permission_cls.objects.exists() or rd_cls.objects.exists():
+        logger.info('Running DABContentType creation script as part of 0005 migration')
+        from ansible_base.rbac.management.create_types import create_DAB_contenttypes
+
+        create_DAB_contenttypes(apps=apps)

ansible_base/resource_registry/models/service_identifier.py

@@ -1,6 +1,8 @@
+import sys
 import uuid
 
-from django.db import models
+from django.conf import settings
+from django.db import IntegrityError, models, transaction
 
 
 class ServiceID(models.Model):
@@ -23,5 +25,21 @@ def save(self, *args, **kwargs):
 def service_id():
     global _service_id
     if not _service_id:
-        _service_id = str(ServiceID.objects.first().pk)
+        obj = ServiceID.objects.first()
+        if obj is None:
+            if settings.DEBUG or "pytest" in sys.argv:
+                try:
+                    with transaction.atomic():
+                        obj = ServiceID.objects.create()
+                        # Check if another process also created one during the race
+                        if ServiceID.objects.count() > 1:
+                            # We lost the race, delete ours and use the other
+                            obj.delete()
+                            obj = ServiceID.objects.first()
+                except IntegrityError:
+                    # Another thread/process won the race—read it
+                    obj = ServiceID.objects.first()
+            else:
+                raise RuntimeError('Expected ServiceID to be created in data migrations but was not found')
+        _service_id = str(obj.pk)
     return _service_id