Improve plugin authentication

CHANGELOG.md

@@ -5,6 +5,12 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## Unreleased
+
+### Fixed
+
+- Improve plugin authentication by @vadimkerr ([#1995](https://github.com/grafana/oncall/pull/1995))
+
 ## v1.2.27 (2023-05-23)
 
 ### Added

engine/apps/auth_token/auth.py

@@ -72,7 +72,14 @@ def authenticate_credentials(self, token_string: str, request: Request) -> Tuple
         if not context_string:
             raise exceptions.AuthenticationFailed("No instance context provided.")
 
-        context = json.loads(context_string)
+        try:
+            context = dict(json.loads(context_string))
+        except (ValueError, TypeError):
+            raise exceptions.AuthenticationFailed("Instance context must be JSON dict.")
+
+        if "stack_id" not in context or "org_id" not in context:
+            raise exceptions.AuthenticationFailed("Invalid instance context.")
+
         try:
             auth_token = check_token(token_string, context=context)
             if not auth_token.organization:
@@ -85,11 +92,19 @@ def authenticate_credentials(self, token_string: str, request: Request) -> Tuple
 
     @staticmethod
     def _get_user(request: Request, organization: Organization) -> User:
-        context = json.loads(request.headers.get("X-Grafana-Context"))
+        try:
+            context = dict(json.loads(request.headers.get("X-Grafana-Context")))
+        except (ValueError, TypeError):
+            raise exceptions.AuthenticationFailed("Grafana context must be JSON dict.")
+
+        if "UserId" not in context and "UserID" not in context:
+            raise exceptions.AuthenticationFailed("Invalid Grafana context.")
+
         try:
             user_id = context["UserId"]
         except KeyError:
             user_id = context["UserID"]
+
         try:
             return organization.users.get(user_id=user_id)
         except User.DoesNotExist:

engine/apps/auth_token/models/plugin_auth_token.py

@@ -1,6 +1,6 @@
 import binascii
 from hmac import compare_digest
-from typing import Optional, Tuple
+from typing import Tuple
 
 from django.db import models
 
@@ -38,7 +38,7 @@ def create_auth_token(cls, organization: Organization) -> Tuple["PluginAuthToken
         return auth_token, token_string
 
     @classmethod
-    def validate_token_string(cls, token: str, *args, **kwargs) -> Optional["PluginAuthToken"]:
+    def validate_token_string(cls, token: str, *args, **kwargs) -> "PluginAuthToken":
         context = kwargs["context"]
         for auth_token in cls.objects.filter(token_key=token[: constants.TOKEN_KEY_LENGTH]):
             try:
@@ -51,3 +51,5 @@ def validate_token_string(cls, token: str, *args, **kwargs) -> Optional["PluginA
                 raise InvalidToken
             if compare_digest(digest, auth_token.digest) and token == recreated_token:
                 return auth_token
+
+        raise InvalidToken

engine/apps/auth_token/tests/test_plugin_auth.py

@@ -0,0 +1,77 @@
+import pytest
+from django.utils import timezone
+from rest_framework.exceptions import AuthenticationFailed
+from rest_framework.test import APIRequestFactory
+
+from apps.auth_token.auth import PluginAuthentication
+
+
+@pytest.mark.django_db
+def test_plugin_authentication_self_hosted_success(make_organization, make_user, make_token_for_organization):
+    organization = make_organization(stack_id=42, org_id=24)
+    user = make_user(organization=organization, user_id=12)
+    token, token_string = make_token_for_organization(organization)
+
+    headers = {
+        "HTTP_AUTHORIZATION": token_string,
+        "HTTP_X-Instance-Context": '{"stack_id": 42, "org_id": 24}',
+        "HTTP_X-Grafana-Context": '{"UserId": 12}',
+    }
+    request = APIRequestFactory().get("/", **headers)
+
+    assert PluginAuthentication().authenticate(request) == (user, token)
+
+
+@pytest.mark.django_db
+def test_plugin_authentication_gcom_success(make_organization, make_user, make_token_for_organization):
+    # Setting gcom_token_org_last_time_synced to now, so it doesn't try to sync with gcom
+    organization = make_organization(
+        stack_id=42, org_id=24, gcom_token="123", gcom_token_org_last_time_synced=timezone.now()
+    )
+    user = make_user(organization=organization, user_id=12)
+
+    headers = {
+        "HTTP_AUTHORIZATION": "gcom:123",
+        "HTTP_X-Instance-Context": '{"stack_id": 42, "org_id": 24}',
+        "HTTP_X-Grafana-Context": '{"UserId": 12}',
+    }
+    request = APIRequestFactory().get("/", **headers)
+
+    ret_user, ret_token = PluginAuthentication().authenticate(request)
+    assert ret_user == user
+    assert ret_token.organization == organization
+
+
+@pytest.mark.django_db
+@pytest.mark.parametrize("grafana_context", [None, "", "non-json", '"string"', "{}", '{"UserId": 1}'])
+def test_plugin_authentication_fail_grafana_context(
+    make_organization, make_user, make_token_for_organization, grafana_context
+):
+    organization = make_organization(stack_id=42, org_id=24)
+    token, token_string = make_token_for_organization(organization)
+
+    headers = {"HTTP_AUTHORIZATION": token_string, "HTTP_X-Instance-Context": '{"stack_id": 42, "org_id": 24}'}
+    if grafana_context is not None:
+        headers["HTTP_X-Grafana-Context"] = grafana_context
+
+    request = APIRequestFactory().get("/", **headers)
+    with pytest.raises(AuthenticationFailed):
+        PluginAuthentication().authenticate(request)
+
+
+@pytest.mark.django_db
+@pytest.mark.parametrize("authorization", [None, "", "123", "gcom:123"])
+@pytest.mark.parametrize("instance_context", [None, "", "non-json", '"string"', "{}", '{"stack_id": 1, "org_id": 1}'])
+def test_plugin_authentication_fail(authorization, instance_context):
+    headers = {}
+
+    if authorization is not None:
+        headers["HTTP_AUTHORIZATION"] = authorization
+
+    if instance_context is not None:
+        headers["HTTP_X-Instance-Context"] = instance_context
+
+    request = APIRequestFactory().get("/", **headers)
+
+    with pytest.raises(AuthenticationFailed):
+        PluginAuthentication().authenticate(request)

engine/apps/grafana_plugin/helpers/gcom.py

@@ -20,7 +20,7 @@ def __init__(self, organization):
         self.organization = organization
 
 
-def check_gcom_permission(token_string: str, context) -> Optional["GcomToken"]:
+def check_gcom_permission(token_string: str, context) -> GcomToken:
     """
     Verify that request from plugin is valid. Check it and synchronize the organization details
     with gcom every GCOM_TOKEN_CHECK_PERIOD.
@@ -87,7 +87,7 @@ def check_gcom_permission(token_string: str, context) -> Optional["GcomToken"]:
     return GcomToken(organization)
 
 
-def check_token(token_string: str, context: dict):
+def check_token(token_string: str, context: dict) -> GcomToken | PluginAuthToken:
     token_parts = token_string.split(":")
     if len(token_parts) > 1 and token_parts[0] == "gcom":
         return check_gcom_permission(token_parts[1], context)