Add orcid client service

Author: mtomilov

h/models/user_identity.py

@@ -1,8 +1,14 @@
+from enum import StrEnum
+
 import sqlalchemy as sa
 
 from h.db import Base
 
 
+class ProviderHost(StrEnum):
+    ORCID = "orcid.org"
+
+
 class UserIdentity(Base):
     __tablename__ = "user_identity"
     __table_args__ = (sa.UniqueConstraint("provider", "provider_unique_id"),)

h/schemas/oauth.py

@@ -0,0 +1,74 @@
+import logging
+import secrets
+from typing import Any, ClassVar, TypedDict
+
+from pyramid.request import Request
+
+from h.schemas.base import JSONSchema, ValidationError
+
+logger = logging.getLogger(__name__)
+
+
+class OAuthCallbackSchema(JSONSchema):
+    schema: ClassVar = {  # type: ignore[misc]
+        "type": "object",
+        "required": ["code"],
+        "properties": {
+            "code": {"type": "string"},
+            "state": {"type": "string"},
+            "error": {"type": "string"},
+            "error_description": {"type": "string"},
+        },
+    }
+
+
+class OAuthCallbackData(TypedDict):
+    code: str
+    state: str | None
+    error: str | None
+    error_description: str | None
+
+
+class RetrieveOAuthCallbackSchema:
+    def __init__(self, request: Request) -> None:
+        self._schema = OAuthCallbackSchema()
+        self._request = request
+
+    def validate(self, data: dict[str, Any]) -> OAuthCallbackData:
+        if data.get("state") != self._request.session.pop("oauth2_state", None):
+            raise ValidationError("Invalid oauth state")  # noqa: EM101, TRY003
+
+        return self._schema.validate(data)
+
+    def state_param(self) -> str:
+        state = secrets.token_hex()
+        self._request.session["oauth2_state"] = state
+        return state
+
+
+class OpenIDTokenSchema(JSONSchema):
+    schema: ClassVar = {  # type: ignore[misc]
+        "type": "object",
+        "required": ["access_token", "refresh_token", "expires_in"],
+        "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
+    expires_in: int
+    id_token: str
+
+
+class RetrieveOpenIDTokenSchema:
+    def __init__(self):
+        self._schema = OpenIDTokenSchema()
+
+    def validate(self, data: dict[str, Any]) -> OpenIDTokenData:
+        return self._schema.validate(data)

h/services/__init__.py

@@ -12,9 +12,13 @@
     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.orcid_client import ORCIDClientService
 from h.services.subscription import SubscriptionService
 from h.services.task_done import TaskDoneService
 
@@ -178,3 +182,12 @@ 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
+    )
+    config.register_service_factory(
+        "h.services.orcid_client.factory", iface=ORCIDClientService
+    )

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))

h/services/orcid_client.py

@@ -0,0 +1,87 @@
+import logging
+
+from sqlalchemy.orm import Session
+
+from h.models import User, UserIdentity
+from h.models.user_identity import ProviderHost
+from h.services.jwt import JWTService
+from h.services.openid_client import OpenIDClientService
+from h.services.user import UserService
+
+logger = logging.getLogger(__name__)
+
+
+class ORCIDClientService:
+    def __init__(  # noqa: PLR0913
+        self,
+        session: Session,
+        host: str,
+        client_id: str,
+        client_secret: str,
+        redirect_uri: str,
+        openid_client_service: OpenIDClientService,
+        user_service: UserService,
+    ) -> None:
+        self._session = session
+        self._host = host
+        self._client_id = client_id
+        self._client_secret = client_secret
+        self._redirect_uri = redirect_uri
+        self._openid_client_service = openid_client_service
+        self._user_service = user_service
+
+    def _get_id_token(self, authorization_code: str) -> str:
+        return self._openid_client_service.get_id_token(
+            token_url=self.token_url,
+            redirect_uri=self._redirect_uri,
+            auth=(self._client_id, self._client_secret),
+            authorization_code=authorization_code,
+        )
+
+    def get_orcid(self, authorization_code: str) -> str | None:
+        id_token = self._get_id_token(authorization_code)
+        decoded_id_token = JWTService.decode_token(id_token, self.key_set_url)
+        return decoded_id_token.get("sub")
+
+    def add_identity(self, user: User, orcid: str) -> None:
+        identity = UserIdentity(
+            user=user,
+            provider=ProviderHost.ORCID,
+            provider_unique_id=orcid,
+        )
+        self._session.add(identity)
+
+    @staticmethod
+    def get_identity(user: User) -> UserIdentity | None:
+        for identity in user.identities:
+            if identity.provider == ProviderHost.ORCID:
+                return identity
+        return None
+
+    @property
+    def token_url(self) -> str:
+        return self._api_url("oauth/token")
+
+    @property
+    def key_set_url(self) -> str:
+        return self._api_url("oauth/jwks")
+
+    def orcid_url(self, orcid: str) -> str | None:
+        return self._api_url(orcid) if orcid else None
+
+    def _api_url(self, path: str) -> str:
+        return f"https://{self._host}/{path}"
+
+
+def factory(_context, request) -> ORCIDClientService:
+    settings = request.registry.settings
+
+    return ORCIDClientService(
+        session=request.db,
+        host=settings["orcid_host"],
+        client_id=settings["orcid_client_id"],
+        client_secret=settings["orcid_client_secret"],
+        redirect_uri=request.route_url("orcid.oauth.callback"),
+        openid_client_service=request.find_service(OpenIDClientService),
+        user_service=request.find_service(name="user"),
+    )

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,79 @@
+import pytest
+
+from h.schemas import ValidationError
+from h.schemas.oauth import (
+    RetrieveOAuthCallbackSchema,
+    RetrieveOpenIDTokenSchema,
+)
+
+
+class TestRetrieveOAuthCallbackSchema:
+    def test_validate(self, pyramid_request, schema):
+        data = {"code": "test-code", "state": "test-state"}
+        pyramid_request.session["oauth2_state"] = data["state"]
+
+        result = schema.validate(data)
+
+        assert result == data
+
+    def test_validate_with_invalid_state(self, pyramid_request, schema):
+        data = {"code": "test-code", "state": "test-state"}
+        pyramid_request.session["oauth2_state"] = "different-test-state"
+
+        with pytest.raises(ValidationError, match="Invalid oauth state"):
+            schema.validate(data)
+
+    def test_state_param_generates_token(self, pyramid_request, schema, secrets):
+        state_token = "test-token"  # noqa: S105
+        secrets.token_hex.return_value = state_token
+
+        result = schema.state_param()
+
+        assert result == state_token
+        assert pyramid_request.session["oauth2_state"] == state_token
+
+    @pytest.fixture
+    def schema(self, pyramid_request):
+        return RetrieveOAuthCallbackSchema(pyramid_request)
+
+    @pytest.fixture(autouse=True)
+    def secrets(self, patch):
+        return patch("h.schemas.oauth.secrets")
+
+
+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"},
+                "^'refresh_token' is a required property, 'expires_in' is a required property$",
+            ),
+            (
+                {
+                    "access_token": "test-access-token",
+                    "refresh_token": "test-refresh-token",
+                },
+                "^'expires_in' 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()