Add full api restriction setting (#15656)

Signed-off-by: Yuriy Savinkin <[email protected]>

Author: uber-dendy

awx/api/conf.py

@@ -1,11 +1,17 @@
+# Python
+from functools import lru_cache
+
 # Django
+from django.urls import resolve, URLResolver, URLPattern
+from django.urls.exceptions import Resolver404
 from django.utils.translation import gettext_lazy as _
 
 # Django REST Framework
 from rest_framework import serializers
 
 # AWX
 from awx.conf import fields, register, register_validate
+from awx.settings.defaults import ANONYMOUS_ACCESS_API_ALLOWED_PATHS as DEFAULT_ALLOWED_PATHS
 
 
 register(
@@ -66,11 +72,60 @@
     category_slug='authentication',
 )
 
+register(
+    'RESTRICT_API_ANONYMOUS_ACCESS',
+    field_class=fields.BooleanField,
+    default=False,
+    label=_('Restrict Anonymous API Access'),
+    help_text=_('If true, all API endpoints except those specified in "Allowed URLs for Anonymous Access" will require authentication.'),
+    category=_('Authentication'),
+    category_slug='authentication',
+)
+
+register(
+    'ANONYMOUS_ACCESS_API_ALLOWED_PATHS',
+    field_class=fields.StringListField,
+    default=DEFAULT_ALLOWED_PATHS,
+    label=_('Allowed URLs for Anonymous Access'),
+    help_text=_('A list of API endpoints that can be accessed without authentication, even when "Restrict Anonymous API Access" is enabled.'),
+    category=_('Authentication'),
+    category_slug='authentication',
+)
+
 
 def authentication_validate(serializer, attrs):
     if attrs.get('DISABLE_LOCAL_AUTH', False):
         raise serializers.ValidationError(_("There are no remote authentication systems configured."))
     return attrs
 
 
+@lru_cache(maxsize=128)
+def validate_url_path(path):
+    """Validate and cache the result for a given URL path."""
+    try:
+        resolve(path)
+    except Resolver404:
+        return False
+    return True
+
+
+def allowed_urls_validate(serializer, attrs):
+    '''
+    Validation for allowed URLs in ANONYMOUS_ACCESS_API_ALLOWED_PATHS
+    This ensures that administrators provide resolvable URLs
+    and include the required default URLs for core functionality.
+    '''
+    paths = attrs.get('ANONYMOUS_ACCESS_API_ALLOWED_PATHS', [])
+    invalid_paths = [path for path in paths if not validate_url_path(path)]
+
+    if invalid_paths:
+        raise serializers.ValidationError(_(f"Invalid paths: {', '.join(invalid_paths)}"))
+
+    missing_paths = [path for path in DEFAULT_ALLOWED_PATHS if path not in paths]
+    if missing_paths:
+        attrs['ANONYMOUS_ACCESS_API_ALLOWED_PATHS'] = missing_paths + paths
+    return attrs
+
+
 register_validate('authentication', authentication_validate)
+register_validate('authentication', allowed_urls_validate)

awx/conf/tests/unit/test_validators.py

@@ -0,0 +1,23 @@
+import pytest
+from rest_framework import serializers
+
+from awx.api.conf import allowed_urls_validate
+from awx.settings.defaults import ANONYMOUS_ACCESS_API_ALLOWED_PATHS as DEFAULT_ALLOWED_PATHS
+
+
+class TestAllowedUrlsValidator:
+
+    def test_ok_validator(self):
+        attrs = {'ANONYMOUS_ACCESS_API_ALLOWED_PATHS': DEFAULT_ALLOWED_PATHS}
+        allowed_urls_validate(None, attrs)
+
+    def test_all_validator(self):
+        attrs = {'ANONYMOUS_ACCESS_API_ALLOWED_PATHS': []}
+        allowed_urls_validate(None, attrs)
+        assert attrs.get('ANONYMOUS_ACCESS_API_ALLOWED_PATHS') == DEFAULT_ALLOWED_PATHS
+
+    def test_wrong_path_validator(self):
+        attrs = {'ANONYMOUS_ACCESS_API_ALLOWED_PATHS': DEFAULT_ALLOWED_PATHS + ['not_a_path']}
+
+        with pytest.raises(serializers.ValidationError):
+            allowed_urls_validate(None, attrs)

awx/main/middleware.py

@@ -12,9 +12,12 @@
 from django.contrib.auth import logout
 from django.db.migrations.recorder import MigrationRecorder
 from django.db import connection
+from django.http import JsonResponse
 from django.shortcuts import redirect
 from django.utils.deprecation import MiddlewareMixin
+from django.utils.translation import gettext_lazy as _
 from django.urls import reverse, resolve
+from rest_framework import status
 
 from awx.main import migrations
 from awx.main.utils.profiling import AWXProfiler
@@ -223,3 +226,29 @@ def process_request(self, request):
             request.urlconf = self._url_optional(prefix)
         else:
             request.urlconf = 'awx.urls'
+
+
+class AnonymousAccessRestrictionMiddleware(MiddlewareMixin):
+    """
+    Restrict anonymous access to API endpoints if RESTRICT_API_ANONYMOUS_ACCESS = True,
+    except for those listed in the ANONYMOUS_ACCESS_API_ALLOWED_PATHS.
+    If access is restricted, unauthenticated users will receive a 401 Unauthorized response.
+    """
+
+    @functools.lru_cache
+    def get_allowed_paths(self):
+        return set(settings.ANONYMOUS_ACCESS_API_ALLOWED_PATHS)
+
+    def process_request(self, request):
+        if not request.path.startswith('/api'):
+            return
+
+        if settings.RESTRICT_API_ANONYMOUS_ACCESS and request.path not in self.get_allowed_paths():
+            if not request.user.is_authenticated:
+                msg = _('Anonymous API access restricted.') + _(' To establish a login session, visit') + ' /api/login/.'
+                return JsonResponse(
+                    {
+                        'detail': msg,
+                    },
+                    status=status.HTTP_401_UNAUTHORIZED,
+                )

awx/main/tests/unit/test_middleware.py

@@ -0,0 +1,67 @@
+import pytest
+from unittest.mock import patch
+
+from django.contrib.auth.models import AnonymousUser
+from django.http import JsonResponse, HttpResponse
+from rest_framework.test import APIRequestFactory
+
+from awx.main.middleware import AnonymousAccessRestrictionMiddleware
+
+
+@pytest.fixture
+def access_restriction_middleware():
+    return AnonymousAccessRestrictionMiddleware(lambda request: HttpResponse())
+
+
+@pytest.fixture
+def mock_user(is_authenticated):
+    return type("User", (), {"is_authenticated": is_authenticated})()
+
+
+class TestAnonymousAccessRestrictionMiddleware:
+    @pytest.mark.parametrize(
+        "is_authenticated,expected_response",
+        [
+            (False, JsonResponse),  # Anonymous user, restricted path
+            (True, None),  # Authenticated user, not restricted
+        ],
+    )
+    @patch("django.conf.settings.RESTRICT_API_ANONYMOUS_ACCESS", True)
+    @patch("django.conf.settings.ANONYMOUS_ACCESS_API_ALLOWED_PATHS", ["/api/public"])
+    def test_restricted_access_to_authenticated_only_path(self, access_restriction_middleware, mock_user, is_authenticated, expected_response):
+        request = APIRequestFactory().get("/api/secure-data")
+        request.user = mock_user
+        response = access_restriction_middleware.process_request(request)
+
+        if expected_response:
+            assert isinstance(response, expected_response)
+            assert response.status_code == 401
+        else:
+            assert response is None
+
+    @patch("django.conf.settings.RESTRICT_API_ANONYMOUS_ACCESS", True)
+    @patch("django.conf.settings.ANONYMOUS_ACCESS_API_ALLOWED_PATHS", ["/api/public"])
+    def test_allowed_path_for_anonymous_user(self, access_restriction_middleware):
+        """Test that anonymous users can access paths in the allowed list."""
+        request = APIRequestFactory().get("/api/public")
+        request.user = AnonymousUser()
+
+        response = access_restriction_middleware.process_request(request)
+        assert response is None
+
+    @patch("django.conf.settings.RESTRICT_API_ANONYMOUS_ACCESS", False)
+    def test_anonymous_access_when_restriction_disabled(self, access_restriction_middleware):
+        """Test that anonymous access is allowed when the restriction is disabled."""
+        request = APIRequestFactory().get("/api/secure-data")
+        request.user = AnonymousUser()  # Anonymous user
+
+        response = access_restriction_middleware.process_request(request)
+        assert response is None
+
+    def test_non_api_path_is_skipped(self, access_restriction_middleware):
+        """Test that non-API paths are skipped by the middleware."""
+        request = APIRequestFactory().get("/")
+        request.user = AnonymousUser()
+
+        response = access_restriction_middleware.process_request(request)
+        assert response is None

awx/settings/defaults.py

@@ -893,6 +893,7 @@
     'crum.CurrentRequestUserMiddleware',
     'awx.main.middleware.URLModificationMiddleware',
     'awx.main.middleware.SessionTimeoutMiddleware',
+    'awx.main.middleware.AnonymousAccessRestrictionMiddleware',
 ]
 
 # Secret header value to exchange for websockets responsible for distributing websocket messages.
@@ -1057,3 +1058,9 @@
 
 # feature flags
 FLAGS = {}
+
+# Restrict access to all endpoints with exception for ANONYMOUS_ACCESS_API_ALLOWED_PATHS
+RESTRICT_API_ANONYMOUS_ACCESS = False
+
+# These are the essential endpoints required for authentication and token management for RESTRICT_API_ANONYMOUS_ACCESS.
+ANONYMOUS_ACCESS_API_ALLOWED_PATHS = ['/api/login/']