add token support for basic auth (wip) (#1999)

CHANGELOG.rst

@@ -14,6 +14,8 @@ Added
 - **General**
     - Python v3.11 support (#1922, #1978)
     - ``SESSION_COOKIE_AGE`` and ``SESSION_EXPIRE_AT_BROWSER_CLOSE`` Django settings (#2015)
+- **Irodsbackend**
+    - Add token auth support to ``BasicAuthAPIView`` (#1999)
 - **Landingzones**
     - REST API list view pagination (#1994)
     - ``notify_email_zone_status`` user app setting (#1939)
@@ -38,6 +40,8 @@ Changed
     - Upgrade to python-irodsclient v2.1.0 (#2007)
     - Upgrade minimum supported iRODS version to v4.3.3 (#1815, #2007)
     - Use constants for timeline event status types (#2010)
+- **Irodsbackend**
+    - Rename ``LocalAuthAPIView`` to ``BasicAuthAPIView`` (#1999)
 - **Irodsinfo**
     - Update REST API versioning (#1936)
 - **Landingzones**

irodsbackend/tests/test_views.py

@@ -7,14 +7,22 @@
 
 from test_plus.test import TestCase
 
+from irodsbackend.views import TOKEN_USER_NAME
+
+# Projectroles dependency
+from projectroles.tests.test_views_api import (
+    SODARAPIViewTestMixin,
+    EMPTY_KNOX_TOKEN,
+)
+
 
 # Local constants
 LOCAL_USER_NAME = 'local_user'
 LOCAL_USER_PASS = 'password'
 
 
-class TestLocalAuthAPIView(TestCase):
-    """Tests for LocalAuthAPIView"""
+class TestBasicAuthAPIView(SODARAPIViewTestMixin, TestCase):
+    """Tests for BasicAuthAPIView"""
 
     @staticmethod
     def _get_auth_header(username, password):
@@ -28,36 +36,58 @@ def _get_auth_header(username, password):
 
     def setUp(self):
         self.user = self.make_user(LOCAL_USER_NAME, LOCAL_USER_PASS)
+        self.url = reverse('irodsbackend:api_auth')
 
-    def test_auth(self):
-        """Test auth with existing user and auth check enabled"""
+    def test_post(self):
+        """Test TestBasicAuthAPIView POST with existing local user"""
         response = self.client.post(
-            reverse('irodsbackend:api_auth'),
-            **self._get_auth_header(LOCAL_USER_NAME, LOCAL_USER_PASS)
+            self.url, **self._get_auth_header(LOCAL_USER_NAME, LOCAL_USER_PASS)
         )
         self.assertEqual(response.status_code, 200)
 
     @override_settings(IRODS_SODAR_AUTH=False)
-    def test_auth_disabled(self):
-        """Test auth with existing user and auth check disabled"""
+    def test_post_disabled(self):
+        """Test POST with local and auth check disabled"""
         response = self.client.post(
-            reverse('irodsbackend:api_auth'),
-            **self._get_auth_header(LOCAL_USER_NAME, LOCAL_USER_PASS)
+            self.url, **self._get_auth_header(LOCAL_USER_NAME, LOCAL_USER_PASS)
         )
         self.assertEqual(response.status_code, 500)
 
-    def test_auth_invalid_user(self):
-        """Test auth with invalid user"""
+    def test_post_invalid_user(self):
+        """Test POST with invalid user"""
         response = self.client.post(
-            reverse('irodsbackend:api_auth'),
+            self.url,
             **self._get_auth_header(LOCAL_USER_NAME, 'invalid_password')
         )
         self.assertEqual(response.status_code, 401)
 
-    def test_auth_invalid_password(self):
-        """Test auth with invalid password"""
+    def test_post_invalid_password(self):
+        """Test POST with invalid password"""
+        response = self.client.post(
+            self.url, **self._get_auth_header('invalid_user', LOCAL_USER_PASS)
+        )
+        self.assertEqual(response.status_code, 401)
+
+    def test_post_token(self):
+        """Test POST with knox token"""
+        knox_token = self.get_token(self.user)
+        response = self.client.post(
+            self.url, **self._get_auth_header(TOKEN_USER_NAME, knox_token)
+        )
+        self.assertEqual(response.status_code, 200)
+
+    def test_post_token_invalid(self):
+        """Test POST with invalid knox token (should fail)"""
+        self.get_token(self.user)  # Making sure the user has A token
+        response = self.client.post(
+            self.url, **self._get_auth_header(TOKEN_USER_NAME, EMPTY_KNOX_TOKEN)
+        )
+        self.assertEqual(response.status_code, 401)
+
+    def test_post_token_username(self):
+        """Test POST with knox token and regular username (should fail)"""
+        knox_token = self.get_token(self.user)
         response = self.client.post(
-            reverse('irodsbackend:api_auth'),
-            **self._get_auth_header('invalid_user', LOCAL_USER_PASS)
+            self.url, **self._get_auth_header(LOCAL_USER_NAME, knox_token)
         )
         self.assertEqual(response.status_code, 401)

irodsbackend/urls.py

@@ -18,7 +18,7 @@
     ),
     path(
         route='api/auth',
-        view=views.LocalAuthAPIView.as_view(),
+        view=views.BasicAuthAPIView.as_view(),
         name='api_auth',
     ),
 ]

irodsbackend/views.py

@@ -1,21 +1,22 @@
 """Views for the irodsbackend app"""
 
+import base64
 import logging
 
 from django.conf import settings
+from django.contrib.auth import authenticate
 from django.http import JsonResponse
+from django.views.generic import View
 
-from rest_framework.exceptions import NotAuthenticated
 from rest_framework.response import Response
-from rest_framework.views import APIView
+
+from knox.auth import TokenAuthentication
 
 # Projectroles dependency
 from projectroles.models import SODAR_CONSTANTS
 from projectroles.plugins import get_backend_api
 from projectroles.views_ajax import SODARBaseProjectAjaxView
 
-from sodar.users.auth import fallback_to_auth_basic
-
 
 logger = logging.getLogger(__name__)
 
@@ -32,6 +33,9 @@
     'Unable to initialize omics_irods backend, iRODS server '
     'possibly unavailable'
 )
+BASIC_AUTH_LOG_PREFIX = 'Basic auth'
+BASIC_AUTH_NOT_ENABLED_MSG = 'IRODS_SODAR_AUTH not enabled'
+TOKEN_USER_NAME = '__token__'
 
 
 class BaseIrodsAjaxView(SODARBaseProjectAjaxView):
@@ -223,30 +227,58 @@ def get(self, request, *args, **kwargs):
             return Response(self._get_detail(ex), status=500)
 
 
-@fallback_to_auth_basic
-class LocalAuthAPIView(APIView):
+# TODO: Make this into a reusable base class/mixin to also use with IGV
+# TODO: Standardize hacks as good as possible (standard responses etc)
+class BasicAuthAPIView(View):
     """
-    REST API view for verifying login credentials for local users in iRODS.
+    REST API view for verifying login credentials for OIDC and local users in
+    iRODS. Does not log in the user.
 
-    Should only be used in local development and testing situations or when an
-    external LDAP/AD login is not available.
+    To be used in environments enabling OIDC access and/or where external
+    LDAP/AD login is not available.
     """
 
     def post(self, request, *args, **kwargs):
         # TODO: Limit access to iRODS host?
-        log_prefix = 'Local auth'
         if not settings.IRODS_SODAR_AUTH:
-            not_enabled_msg = 'IRODS_SODAR_AUTH not enabled'
-            logger.error('{} failed: {}'.format(log_prefix, not_enabled_msg))
-            return JsonResponse({'detail': not_enabled_msg}, status=500)
-        if request.user.is_authenticated:
+            logger.error(
+                '{} failed: {}'.format(
+                    BASIC_AUTH_LOG_PREFIX, BASIC_AUTH_NOT_ENABLED_MSG
+                )
+            )
+            return JsonResponse(
+                {'detail': BASIC_AUTH_NOT_ENABLED_MSG}, status=500
+            )
+        if 'HTTP_AUTHORIZATION' not in request.META:
+            return JsonResponse(
+                {'detail': 'Auth header not included'}, status=400
+            )
+        user = None
+        auth = request.META['HTTP_AUTHORIZATION'].split()
+        if len(auth) == 2 and auth[0].lower() == 'basic':
+            uname, passwd = base64.b64decode(auth[1]).decode().split(':')
+            # For token user, auth against Knox token
+            if uname == TOKEN_USER_NAME:
+                token_auth = TokenAuthentication()
+                try:
+                    user, _ = token_auth.authenticate_credentials(
+                        passwd.encode('utf-8')
+                    )
+                except Exception as ex:
+                    logger.error('Token auth failed: {}'.format(ex))
+            else:  # For local user, do standard password auth
+                user = authenticate(username=uname, password=passwd)
+        if user and user.is_authenticated:
             logger.info(
-                '{} successful: {}'.format(log_prefix, request.user.username)
+                '{} successful: {}'.format(
+                    BASIC_AUTH_LOG_PREFIX, request.user.username
+                )
             )
             return JsonResponse({'detail': 'ok'}, status=200)
         logger.error(
             '{} failed: User {} not authenticated'.format(
-                log_prefix, request.user.username
+                BASIC_AUTH_LOG_PREFIX, request.user.username
             )
         )
-        raise NotAuthenticated()
+        # TODO: Return proper response
+        return JsonResponse({'detail': 'Unauthorized'}, status=401)