Add openid client service

Author: mtomilov

h/schemas/oauth.py

@@ -0,0 +1,34 @@
+import logging
+from typing import Any, ClassVar, TypedDict
+
+from h.schemas.base import JSONSchema
+
+logger = logging.getLogger(__name__)
+
+
+class OpenIDTokenSchema(JSONSchema):
+    schema: ClassVar = {  # type: ignore[misc]
+        "type": "object",
+        "required": ["access_token", "expires_in", "id_token"],
+        "properties": {
+            "access_token": {"type": "string"},
+            "refresh_token": {"type": "string"},
+            "expires_in": {"type": "integer", "minimum": 1},
+            "id_token": {"type": "string"},
+        },
+    }
+
+
+class OpenIDTokenData(TypedDict):
+    access_token: str
+    refresh_token: str | None
+    expires_in: int
+    id_token: str
+
+
+class RetrieveOpenIDTokenSchema:
+    def __init__(self) -> None:
+        self._schema = OpenIDTokenSchema()
+
+    def validate(self, data: dict[str, Any]) -> OpenIDTokenData:
+        return self._schema.validate(data)

h/services/__init__.py

@@ -12,9 +12,12 @@
     BulkLMSStatsService,
 )
 from h.services.email import EmailService
+from h.services.http import HTTPService
 from h.services.job_queue import JobQueueService
+from h.services.jwt import JWTService
 from h.services.mention import MentionService
 from h.services.notification import NotificationService
+from h.services.openid_client import OpenIDClientService
 from h.services.subscription import SubscriptionService
 from h.services.task_done import TaskDoneService
 
@@ -178,3 +181,9 @@ def includeme(config):  # pragma: no cover  # noqa: PLR0915
     config.register_service_factory(
         "h.services.task_done.factory", iface=TaskDoneService
     )
+
+    # Authentication-related services
+    config.register_service_factory("h.services.http.factory", iface=HTTPService)
+    config.register_service_factory(
+        "h.services.openid_client.factory", iface=OpenIDClientService
+    )

h/services/openid_client.py

@@ -0,0 +1,51 @@
+import logging
+from typing import Any
+
+from h.schemas.oauth import OpenIDTokenData, RetrieveOpenIDTokenSchema
+from h.services.http import ExternalRequestError, HTTPService
+
+logger = logging.getLogger(__name__)
+
+
+class OpenIDTokenError(ExternalRequestError):
+    """
+    A problem with an Open ID token for an external API.
+
+    This is raised when we don't have an access token for the current user or
+    when our access token doesn't work (e.g. because it's expired or been
+    revoked).
+    """
+
+
+class OpenIDClientService:
+    def __init__(self, http_service: HTTPService) -> None:
+        self._http_service = http_service
+
+    def get_id_token(
+        self,
+        token_url: str,
+        redirect_uri: str,
+        auth: tuple[str, str],
+        authorization_code: str,
+    ) -> str:
+        data = self._request_openid_data(
+            token_url=token_url,
+            auth=auth,
+            data={
+                "redirect_uri": redirect_uri,
+                "grant_type": "authorization_code",
+                "code": authorization_code,
+            },
+        )
+        return data["id_token"]
+
+    def _request_openid_data(
+        self, token_url: str, data: dict[str, Any], auth: tuple[str, str]
+    ) -> OpenIDTokenData:
+        response = self._http_service.post(token_url, data=data, auth=auth)
+
+        return RetrieveOpenIDTokenSchema().validate(response.json())
+
+
+def factory(_context, request) -> OpenIDClientService:
+    return OpenIDClientService(request.find_service(HTTPService))

tests/common/fixtures/services.py

@@ -2,7 +2,12 @@
 
 import pytest
 
-from h.services import MentionService, NotificationService
+from h.services import (
+    HTTPService,
+    MentionService,
+    NotificationService,
+    OpenIDClientService,
+)
 from h.services.analytics import AnalyticsService
 from h.services.annotation_authority_queue import AnnotationAuthorityQueueService
 from h.services.annotation_delete import AnnotationDeleteService
@@ -74,6 +79,7 @@
     "group_members_service",
     "group_service",
     "group_update_service",
+    "http_service",
     "links_service",
     "list_organizations_service",
     "mention_service",
@@ -82,6 +88,7 @@
     "nipsa_service",
     "notification_service",
     "oauth_provider_service",
+    "openid_client_service",
     "organization_service",
     "queue_service",
     "search_index",
@@ -347,3 +354,13 @@ def email_service(mock_service):
 @pytest.fixture
 def task_done_service(mock_service):
     return mock_service(TaskDoneService)
+
+
+@pytest.fixture
+def http_service(mock_service):
+    return mock_service(HTTPService)
+
+
+@pytest.fixture
+def openid_client_service(mock_service):
+    return mock_service(OpenIDClientService)

tests/unit/h/schemas/oauth_test.py

@@ -0,0 +1,42 @@
+import pytest
+
+from h.schemas import ValidationError
+from h.schemas.oauth import RetrieveOpenIDTokenSchema
+
+
+class TestRetrieveOpenIDTokenSchema:
+    def test_validate(self, schema):
+        data = {
+            "access_token": "test_access_token",
+            "refresh_token": "test_refresh_token",
+            "expires_in": 3600,
+            "id_token": "test_id_token",
+        }
+
+        result = schema.validate(data)
+
+        assert result == data
+
+    @pytest.mark.parametrize(
+        "data,expected_error",
+        [
+            (
+                {"access_token": "test_access_token"},
+                "^'expires_in' is a required property, 'id_token' is a required property$",
+            ),
+            (
+                {
+                    "access_token": "test_access_token",
+                    "expires_in": 3600,
+                },
+                "^'id_token' is a required property$",
+            ),
+        ],
+    )
+    def test_validate_with_invalid_data(self, data, expected_error, schema):
+        with pytest.raises(ValidationError, match=expected_error):
+            schema.validate(data)
+
+    @pytest.fixture
+    def schema(self):
+        return RetrieveOpenIDTokenSchema()

tests/unit/h/services/openid_client_test.py

@@ -0,0 +1,82 @@
+from unittest.mock import sentinel
+
+import pytest
+
+from h.services.exceptions import ExternalRequestError
+from h.services.openid_client import OpenIDClientService, factory
+
+
+class TestOpenIDClientService:
+    def test_get_id_token(
+        self, svc, http_service, passed_args, RetrieveOpenIDTokenSchema, id_token_data
+    ):
+        http_response = http_service.post.return_value
+        http_response.json.return_value = id_token_data
+        RetrieveOpenIDTokenSchema.return_value.validate.return_value = id_token_data
+
+        result = svc.get_id_token(**passed_args)
+
+        http_service.post.assert_called_once_with(
+            passed_args["token_url"],
+            data={
+                "redirect_uri": passed_args["redirect_uri"],
+                "grant_type": "authorization_code",
+                "code": passed_args["authorization_code"],
+            },
+            auth=passed_args["auth"],
+        )
+        RetrieveOpenIDTokenSchema.return_value.validate.assert_called_once_with(
+            http_response.json.return_value
+        )
+        assert result == id_token_data["id_token"]
+
+    def test_get_id_token_raises_if_the_request_fails(
+        self, svc, http_service, passed_args
+    ):
+        http_service.post.side_effect = ExternalRequestError(
+            request=sentinel.err_request, response=sentinel.err_response
+        )
+
+        with pytest.raises(ExternalRequestError) as exc_info:
+            svc.get_id_token(**passed_args)
+
+        assert exc_info.value.request == sentinel.err_request
+        assert exc_info.value.response == sentinel.err_response
+
+    @pytest.fixture
+    def id_token_data(self):
+        return {
+            "access_token": "test_access_token",
+            "refresh_token": "test_refresh_token",
+            "expires_in": 3600,
+            "id_token": "test_id_token",
+        }
+
+    @pytest.fixture
+    def passed_args(self):
+        return {
+            "token_url": sentinel.token_url,
+            "redirect_uri": sentinel.redirect_uri,
+            "auth": sentinel.auth,
+            "authorization_code": sentinel.authorization_code,
+        }
+
+    @pytest.fixture
+    def svc(self, http_service):
+        return OpenIDClientService(http_service)
+
+    @pytest.fixture(autouse=True)
+    def RetrieveOpenIDTokenSchema(self, patch):
+        return patch("h.services.openid_client.RetrieveOpenIDTokenSchema")
+
+
+class TestFactory:
+    def test_it(self, pyramid_request, http_service, OpenIDClientService):
+        svc = factory(sentinel.context, pyramid_request)
+
+        OpenIDClientService.assert_called_once_with(http_service)
+        assert svc == OpenIDClientService.return_value
+
+    @pytest.fixture(autouse=True)
+    def OpenIDClientService(self, patch):
+        return patch("h.services.openid_client.OpenIDClientService")