[RR] Obtain an auth token given user/pw

Signed-off-by: Rick Elrod <[email protected]>

ansible_base/lib/testing/fixtures.py

@@ -1,3 +1,4 @@
+import copy
 import os
 import uuid
 from collections import namedtuple
@@ -42,6 +43,33 @@ def _env(key, value):
     return _env
 
 
+@pytest.fixture
+def settings_override_mutable(settings):
+    """
+    pytest-django's settings  doesn't handle mutable settings types well
+    (https://github.com/pytest-dev/pytest-django/issues/601). This fixture is an
+    attempt to solve that problem.
+
+    Use as follows:
+
+    def test_foo(settings, settings_override_mutable):
+        with settings_override_mutable("SOME_SETTING_VAR"):
+            settings.SOME_SETTING_VAR['and']['nested']['a']['bit'] = "new_value"
+            # or even
+            delattr(settings, 'SOME_SETTING_VAR')  # for some reason normal "del" doesn't work here
+        # But out of the context manager the whole var is restored
+        assert settings.SOME_SETTING_VAR['and']['nested']['a']['bit'] == "old_value"
+    """
+
+    @contextmanager
+    def f(setting_key):
+        original = copy.deepcopy(getattr(settings, setting_key))
+        yield
+        setattr(settings, setting_key, original)
+
+    return f
+
+
 @pytest.fixture
 def local_authenticator(db):
     from ansible_base.authentication.models import Authenticator

ansible_base/resource_registry/views.py

@@ -7,6 +7,7 @@
 from rest_framework import permissions
 from rest_framework.decorators import action
 from rest_framework.pagination import PageNumberPagination
+from rest_framework.permissions import AllowAny
 from rest_framework.response import Response
 from rest_framework.viewsets import GenericViewSet, mixins
 
@@ -15,6 +16,7 @@
 from ansible_base.resource_registry.models import Resource, ResourceType, service_id
 from ansible_base.resource_registry.registry import get_registry
 from ansible_base.resource_registry.serializers import ResourceListSerializer, ResourceSerializer, ResourceTypeSerializer, UserAuthenticationSerializer
+from ansible_base.resource_registry.utils.auth_code import get_user_auth_code
 from ansible_base.rest_filters.rest_framework.field_lookup_backend import FieldLookupBackend
 from ansible_base.rest_filters.rest_framework.order_backend import OrderByBackend
 from ansible_base.rest_filters.rest_framework.type_filter_backend import TypeFilterBackend
@@ -173,12 +175,10 @@ class ValidateLocalUserView(AnsibleBaseDjangoAppApiView):
     Validate a user's username and password.
     """
 
-    permission_classes = [
-        HasResourceRegistryPermissions,
-    ]
-
     custom_action_label = "validate-local-user"
 
+    permission_classes = [AllowAny]
+
     def post(self, request, **kwargs):
         serializer = UserAuthenticationSerializer(data=request.data)
         serializer.is_valid(raise_exception=True)
@@ -189,4 +189,15 @@ def post(self, request, **kwargs):
         if not user:
             return Response(status=401)
 
-        return Response(data={"ansible_id": Resource.get_resource_for_object(user).ansible_id})
+        try:
+            auth_code = get_user_auth_code(user)
+        except AttributeError:
+            logger.exception(f"Cannot generate auth code for user {user}")
+            auth_code = None
+
+        response = {
+            "ansible_id": Resource.get_resource_for_object(user).ansible_id,
+            "auth_code": auth_code,
+        }
+
+        return Response(data=response)

test_app/tests/lib/testing/test_fixtures.py

@@ -0,0 +1,14 @@
+def test_settings_override_mutable(settings_override_mutable, settings):
+    """
+    Ensure that when we modify a mutable setting, it gets reverted.
+    """
+    assert settings.LOGGING['handlers']['console']['formatter'] == "simple"
+
+    with settings_override_mutable('LOGGING'):
+        settings.LOGGING['handlers']['console']['formatter'] = "not so simple"
+        assert settings.LOGGING['handlers']['console']['formatter'] == "not so simple"
+
+        del settings.LOGGING['handlers']['console']['formatter']
+        assert 'formattter' not in settings.LOGGING['handlers']['console']
+
+    assert settings.LOGGING['handlers']['console']['formatter'] == "simple"

test_app/tests/resource_registry/test_resources_api_rest_client.py

@@ -1,10 +1,12 @@
 import uuid
 
+import jwt
 import pytest
 from requests.exceptions import HTTPError
 
 from ansible_base.authentication.models import AuthenticatorUser
 from ansible_base.resource_registry.models import Resource, service_id
+from ansible_base.resource_registry.resource_server import get_resource_server_config
 from ansible_base.resource_registry.rest_client import ResourceAPIClient, ResourceRequestBody
 
 
@@ -187,8 +189,12 @@ def test_validate_local_user(resource_client, admin_user, member_rd):
     resp = resource_client.validate_local_user(username=admin_user.username, password="password")
 
     assert resp.status_code == 200
-    assert resp.json()["ansible_id"] == str(admin_user.resource.ansible_id)
+    json = resp.json()
+    json["ansible_id"] == str(admin_user.resource.ansible_id)
 
-    resp = resource_client.validate_local_user(username=admin_user.username, password="fake password")
+    config = get_resource_server_config()
+    jwt_decoded = jwt.decode(json["auth_code"], config["SECRET_KEY"], config["JWT_ALGORITHM"])
+    assert jwt_decoded['username'] == admin_user.username
 
+    resp = resource_client.validate_local_user(username=admin_user.username, password="fake password")
     assert resp.status_code == 401

test_app/tests/resource_registry/test_views.py

@@ -0,0 +1,22 @@
+from ansible_base.lib.utils.response import get_relative_url
+
+
+def test_validate_local_user(unauthenticated_api_client, admin_user, local_authenticator, settings_override_mutable, settings):
+    url = get_relative_url('validate-local-account')
+    data = {
+        "username": admin_user.username,
+        "password": "password",
+    }
+    response = unauthenticated_api_client.post(url, data=data)
+    assert response.status_code == 200
+    assert 'ansible_id' in response.data
+    assert response.data['auth_code'] is not None
+
+    # If we're missing RESOURCE_SERVER, we can't generate an auth code, so return null instead.
+    with settings_override_mutable('RESOURCE_SERVER'):
+        delattr(settings, 'RESOURCE_SERVER')
+
+        response = unauthenticated_api_client.post(url, data=data)
+        assert response.status_code == 200
+        assert 'ansible_id' in response.data
+        assert response.data['auth_code'] is None

test_app/views.py

@@ -8,7 +8,7 @@
 from rest_framework.response import Response
 from rest_framework.viewsets import ModelViewSet
 
-from ansible_base.lib.utils.response import get_relative_url
+from ansible_base.lib.utils.response import get_fully_qualified_url
 from ansible_base.lib.utils.views.ansible_base import AnsibleBaseView
 from ansible_base.oauth2_provider.views import DABOAuth2UserViewsetMixin
 from ansible_base.rbac import permission_registry
@@ -160,20 +160,20 @@ def api_root(request, format=None):
         # want '^users/$' [name='user-list']
         # do not want '^users/(?P<pk>[^/.]+)/organizations/$' [name='user-organizations-list'],
         if '-list' in url.name and url.pattern._regex.count('/') == 1:
-            list_endpoints[url.name.removesuffix('-list')] = get_relative_url(url.name, request=request, format=format)
+            list_endpoints[url.name.removesuffix('-list')] = get_fully_qualified_url(url.name, request=request, format=format)
 
     from ansible_base.api_documentation.urls import api_version_urls as docs_urls
     from ansible_base.authentication.urls import api_version_urls as authentication_urls
 
     for url in docs_urls + authentication_urls[1:]:
         if isinstance(url, URLPattern):
             try:
-                list_endpoints[url.name] = get_relative_url(url.name, request=request, format=format)
+                list_endpoints[url.name] = get_fully_qualified_url(url.name, request=request, format=format)
             except NoReverseMatch:
                 pass
 
-    list_endpoints['service-index'] = get_relative_url('service-index-root', request=request, format=format)
-    list_endpoints['role-metadata'] = get_relative_url('role-metadata', request=request, format=format)
+    list_endpoints['service-index'] = get_fully_qualified_url('service-index-root', request=request, format=format)
+    list_endpoints['role-metadata'] = get_fully_qualified_url('role-metadata', request=request, format=format)
 
     return Response(list_endpoints)