Commit 177003da authored by Mickaël Desfrênes's avatar Mickaël Desfrênes
Browse files

prevent inactive user from authenticating using API key

parent 145f8437

src/jama/middleware.py

+1 −0
Original line number Diff line number Diff line
@@ -31,4 +31,5 @@ class ApiKeyCsrfViewMiddleware(CsrfViewMiddleware): 
        return APIKey.objects.filter(

            key_hash=APIKey.hash_key(api_key),

            active=True,

            user__is_active=True,

        ).exists()

src/rpc/methods.py

+1 −1
Original line number Diff line number Diff line
@@ -166,7 +166,7 @@ def _require_superuser(func): 
    @_wraps(func)

    def wrapper(*args, **kwargs):

        user = args[0]  # user MUST be first argument

        if not user or not user.is_superuser:

        if not user or not user.is_active or not user.is_superuser:

            raise ServiceException(SUPERUSER_NEEDED)

        return func(*args, **kwargs)

src/rpc/tests.py

+46 −0
Original line number Diff line number Diff line
@@ -17,6 +17,7 @@ from openpyxl import load_workbook 
from unittest.mock import patch

import hashlib

import json

from jama.middleware import ApiKeyCsrfViewMiddleware



object_classes = [

    "collection",
@@ -150,6 +151,40 @@ class ServiceTestCase(TestCase): 


        self.assertIsNone(views._get_user_from_request(request))



    def test_get_user_from_request_rejects_inactive_api_key_user(self):

        self.test_user.is_active = False

        self.test_user.save(update_fields=["is_active"])

        request = self.factory.get("/rpc/", HTTP_X_API_KEY=self.basic_user_key)

        request.user = AnonymousUser()



        self.assertIsNone(views._get_user_from_request(request))



    def test_get_user_from_request_rejects_inactive_session_user(self):

        self.test_user.is_active = False

        self.test_user.save(update_fields=["is_active"])

        request = self.factory.get("/rpc/")

        request.user = self.test_user



        self.assertIsNone(views._get_user_from_request(request))



    def test_get_user_from_request_rejects_inactive_basic_auth_user(self):

        self.test_user.is_active = False

        credentials = base64.b64encode(b"basic_user:password").decode("ascii")

        request = self.factory.get("/rpc/", HTTP_AUTHORIZATION=f"Basic {credentials}")

        request.user = AnonymousUser()



        with patch.object(views, "authenticate", return_value=self.test_user):

            self.assertIsNone(views._get_user_from_request(request))



    def test_csrf_middleware_rejects_api_key_for_inactive_user(self):

        self.test_user.is_active = False

        self.test_user.save(update_fields=["is_active"])

        request = self.factory.post("/rpc/", HTTP_X_API_KEY=self.basic_user_key)



        middleware = ApiKeyCsrfViewMiddleware(lambda req: None)



        self.assertFalse(middleware._request_has_active_api_key(request))



    def test_get_user_from_request_requires_csrf_for_session_post(self):

        request = self.factory.post(

            "/rpc/", data=b"{}", content_type="application/json"
@@ -803,6 +838,17 @@ class ServiceTestCase(TestCase): 
        with self.assertRaises(ServiceException):

            methods.delete_role(self.test_user, 1)



    def test_require_superuser_rejects_inactive_superuser(self):

        self.admin_user.is_active = False

        self.admin_user.save(update_fields=["is_active"])



        with self.assertRaises(ServiceException):

            methods.create_project(self.admin_user, "inactive admin project")



        self.assertFalse(

            models.Project.objects.filter(label="inactive admin project").exists()

        )



    def test_resources_bad_order_by(self):

        collection = methods.add_collection(

            self.test_user,

src/rpc/views.py

+7 −3
Original line number Diff line number Diff line
@@ -195,7 +195,11 @@ def _join_parts( 
def _get_api_key_user_from_request(request: HttpRequest) -> Union[User, None]:

    try:

        key_hash = models.APIKey.hash_key(request.headers["X-Api-Key"])

        return User.objects.get(apikey__key_hash=key_hash, apikey__active=True)

        return User.objects.get(

            apikey__key_hash=key_hash,

            apikey__active=True,

            is_active=True,

        )

    except User.DoesNotExist:

        logger.warning(

            f"X-Api-Key was given but user was not found from ip "
@@ -219,7 +223,7 @@ def _get_basic_auth_user_from_request(request: HttpRequest) -> Union[User, None] 
            logger.warning(

                "Authorization was given in request headers but user was not authenticated"

            )

        return user

        return user if user and user.is_active else None

    except (KeyError, ValueError, UnicodeDecodeError, binascii.Error):

        logger.warning("Authorization was malformed")

        return None
@@ -246,7 +250,7 @@ def _get_user_from_request(request: HttpRequest) -> Union[User, None]: 
        return _get_api_key_user_from_request(request)

    if "Authorization" in request.headers:

        return _get_basic_auth_user_from_request(request)

    if request.user.is_authenticated:

    if request.user.is_authenticated and request.user.is_active:

        if not _request_passes_csrf_check(request):

            logger.warning("CSRF validation failed for session-authenticated request")

            return None