diff --git a/lms/services/canvas_studio.py b/lms/services/canvas_studio.py index add5208425..e94867e9d4 100644 --- a/lms/services/canvas_studio.py +++ b/lms/services/canvas_studio.py @@ -1,11 +1,16 @@ +from functools import lru_cache from typing import Literal, NotRequired, Type, TypedDict from urllib.parse import urlencode, urljoin, urlparse, urlunparse +import requests from marshmallow import EXCLUDE, Schema, fields, post_load +from pyramid.httpexceptions import HTTPBadRequest from lms.models.oauth2_token import Service +from lms.models.user import User from lms.services.aes import AESService from lms.services.exceptions import ExternalRequestError, OAuth2TokenError +from lms.services.oauth_http import OAuthHTTPService from lms.services.oauth_http import factory as oauth_http_factory from lms.validation._base import RequestsResponseSchema @@ -119,6 +124,7 @@ def __init__(self, request, application_instance): {}, request, service=Service.CANVAS_STUDIO ) self._request = request + self._application_instance = application_instance def get_access_token(self, code: str) -> None: """ @@ -256,8 +262,9 @@ def get_canonical_video_url(self, media_id: str) -> str: def get_video_download_url(self, media_id: str) -> str: """Return temporary download URL for a video.""" - download_url = self._api_url(f"v1/media/{media_id}/download") - download_rsp = self._oauth_http_service.get(download_url, allow_redirects=False) + download_rsp = self._bare_api_request( + f"v1/media/{media_id}/download", as_admin=True, allow_redirects=False + ) download_redirect = download_rsp.headers.get("Location") if download_rsp.status_code != 302 or not download_redirect: @@ -276,7 +283,9 @@ def get_transcript_url(self, media_id: str) -> str | None: """ captions = self._api_request( - f"v1/media/{media_id}/caption_files", CanvasStudioCaptionFilesSchema + f"v1/media/{media_id}/caption_files", + CanvasStudioCaptionFilesSchema, + as_admin=True, ) for caption in captions: @@ -286,22 +295,100 @@ def get_transcript_url(self, media_id: str) -> str | None: return None - def _api_request(self, path: str, schema_cls: Type[RequestsResponseSchema]) -> dict: - """Make a request to the Canvas Studio API and parse the JSON response.""" + def _api_request( + self, path: str, schema_cls: Type[RequestsResponseSchema], as_admin=False + ) -> dict: + """ + Make a request to the Canvas Studio API and parse the JSON response. + + :param as_admin: Make the request using the admin account instead of the current user. + """ + response = self._bare_api_request(path, as_admin=as_admin) + return schema_cls(response).parse() + + def _admin_api_request( + self, path: str, allow_redirects=True, allow_refresh=True + ) -> requests.Response: + """ + Make a request to the Canvas Studio API using the admin user identity. + + This method should not be used if the current user _is_ the admin user. + In that case we want to follow the normal steps for making a request + using the current identity, and the corresponding error handling. + """ + + url = self._api_url(path) try: - response = self._oauth_http_service.get(self._api_url(path)) + return self._admin_oauth_http.get(url, allow_redirects=allow_redirects) except ExternalRequestError as err: refreshable = getattr(err.response, "status_code", None) == 401 - if refreshable: - raise OAuth2TokenError( - refreshable=True, - refresh_route="canvas_studio_api.oauth.refresh", - refresh_service=Service.CANVAS_STUDIO, - ) from err + if not refreshable: + # Ordinarily if the access token is missing or expired, we'll + # return an error and the frontend will prompt to + # re-authenticate. That won't help for admin-authenticated + # requests. + if isinstance(err, OAuth2TokenError): + raise HTTPBadRequest( + "The Canvas Studio admin needs to authenticate the Hypothesis integration" + ) from err + raise + + # If we already performed a refresh in response to the original + # request, `allow_refresh` will be False and we abort. + if not allow_refresh: + raise + + # For admin requests, we have to do the refresh here, because the + # refresh path involving the frontend is designed for refreshing + # the current LTI user's access token. + # + # Ideally it would be simpler if we could just encapsulate the + # entire refresh process in OAuthHTTPService. + try: + self._admin_oauth_http.refresh_access_token( + self._token_url(), + self.redirect_uri(), + auth=(self._client_id, self._client_secret), + ) + except ExternalRequestError as refresh_err: + raise ExternalRequestError( + message="Canvas Studio admin token refresh failed. Ask the admin user to re-authenticate." + ) from refresh_err + + # Retry the request with the new token. + return self._admin_api_request( + path, + allow_redirects=allow_redirects, + # If the request fails again, make sure we don't repeat the + # refresh to avoid getting stuck in a loop. + allow_refresh=False, + ) - raise + def _bare_api_request( + self, path: str, as_admin=False, allow_redirects=True + ) -> requests.Response: + """ + Make a request to the Canvas Studio API and return the response. - return schema_cls(response).parse() + :param as_admin: Make the request using the admin account instead of the current user. + """ + url = self._api_url(path) + + if as_admin and not self._is_admin(): + return self._admin_api_request(path, allow_redirects=allow_redirects) + + try: + return self._oauth_http_service.get(url, allow_redirects=allow_redirects) + except ExternalRequestError as err: + refreshable = getattr(err.response, "status_code", None) == 401 + if not refreshable: + raise + + raise OAuth2TokenError( + refreshable=True, + refresh_route="canvas_studio_api.oauth.refresh", + refresh_service=Service.CANVAS_STUDIO, + ) from err def _api_url(self, path: str) -> str: """ @@ -316,6 +403,56 @@ def _api_url(self, path: str) -> str: site = self._canvas_studio_site() return f"{site}/api/public/{path}" + def _admin_email(self) -> str: + """Return the email address of the configured Canvas Studio admin.""" + admin_email = self._application_instance.settings.get( + "canvas_studio", "admin_email" + ) + if not admin_email: + raise HTTPBadRequest( + "Admin account is not configured for Canvas Studio integration" + ) + return admin_email + + def _is_admin(self) -> bool: + """Return true if the current LTI user is the configure Canvas Studio admin.""" + return self._request.lti_user.email == self._admin_email() + + @property + @lru_cache + def _admin_oauth_http(self) -> OAuthHTTPService: + """ + Return an OAuthHTTPService that makes calls using the admin user account. + + Admin accounts have the ability to download all videos and transcripts + in a Canvas Studio instance, whereas videos can ordinarily only be + downloaded by the owner. Therefore when launching Canvas Studio + assignments, we use this account instead of the current user's account + to authenticate the API requests for downloading videos and transcripts. + """ + + # The caller should check for this condition before calling this method + # and use the standard `self._oauth_http_service` property instead. + assert not self._is_admin() + + admin_email = self._admin_email() + admin_user = ( + self._request.db.query(User) + .filter_by( + email=admin_email, application_instance=self._application_instance + ) + .one_or_none() + ) + if not admin_user: + raise HTTPBadRequest( + "The Canvas Studio admin needs to authenticate the Hypothesis integration" + ) + admin_lti_user = admin_user.user_id + + return oauth_http_factory( + {}, self._request, service=Service.CANVAS_STUDIO, user_id=admin_lti_user + ) + def _canvas_studio_site(self) -> str: return f"https://{self._domain}" diff --git a/lms/services/oauth2_token.py b/lms/services/oauth2_token.py index 74a7eb2fea..609a5bcb60 100644 --- a/lms/services/oauth2_token.py +++ b/lms/services/oauth2_token.py @@ -5,7 +5,7 @@ from lms.models import OAuth2Token from lms.models.oauth2_token import Service -from lms.services import OAuth2TokenError +from lms.services.exceptions import OAuth2TokenError class OAuth2TokenService: @@ -69,7 +69,9 @@ def get(self, service=Service.LMS): ) from err -def oauth2_token_service_factory(_context, request): +def oauth2_token_service_factory(_context, request, user_id: str | None = None): return OAuth2TokenService( - request.db, request.lti_user.application_instance, request.lti_user.user_id + request.db, + request.lti_user.application_instance, + user_id or request.lti_user.user_id, ) diff --git a/lms/services/oauth_http.py b/lms/services/oauth_http.py index 99be6e4ca8..2417bfa140 100644 --- a/lms/services/oauth_http.py +++ b/lms/services/oauth_http.py @@ -2,6 +2,7 @@ from lms.models.oauth2_token import Service from lms.services.exceptions import ExternalRequestError, OAuth2TokenError +from lms.services.oauth2_token import oauth2_token_service_factory from lms.validation import RequestsResponseSchema from lms.validation.authentication import OAuthTokenResponseSchema @@ -132,9 +133,26 @@ def _token_request(self, token_url, data, auth): return validated_data["access_token"] -def factory(_context, request, service: Service = Service.LMS) -> OAuthHTTPService: +def factory( + _context, request, service: Service = Service.LMS, user_id: str | None = None +) -> OAuthHTTPService: + """ + Create an `OAuthHTTPService`. + + :param request: The Pyramid request + :param service: The API this service will communicate with + :param user_id: + The LTI user ID of the user whose API tokens should be used. Defaults + to the LTI user from the current request. + """ + if user_id: + oauth2_token_svc = oauth2_token_service_factory( + _context, request, user_id=user_id + ) + else: + oauth2_token_svc = request.find_service(name="oauth2_token") return OAuthHTTPService( request.find_service(name="http"), - request.find_service(name="oauth2_token"), + oauth2_token_svc, service=service, ) diff --git a/tests/unit/lms/services/canvas_studio_test.py b/tests/unit/lms/services/canvas_studio_test.py index cd2e2b9682..79be0fa33f 100644 --- a/tests/unit/lms/services/canvas_studio_test.py +++ b/tests/unit/lms/services/canvas_studio_test.py @@ -3,6 +3,8 @@ import pytest from h_matchers import Any +from pyramid.httpexceptions import HTTPBadRequest +from requests import RequestException from lms.models.oauth2_token import Service from lms.services.canvas_studio import CanvasStudioService, factory @@ -11,7 +13,9 @@ from tests import factories -@pytest.mark.usefixtures("aes_service", "canvas_studio_settings", "oauth_http_factory") +@pytest.mark.usefixtures( + "aes_service", "canvas_studio_settings", "oauth_http_factory", "admin_user" +) class TestCanvasStudioService: def test_get_access_token(self, svc, oauth_http_service, client_secret): @@ -134,13 +138,136 @@ def test_get_video_download_url(self, svc): url = svc.get_video_download_url("42") assert url == "https://videos.cdn.com/video.mp4?signature=abc" - def test_get_video_download_url_error(self, svc): + def test_get_video_download_url_as_admin( + self, + svc, + pyramid_request, + admin_user, + oauth_http_service, + admin_oauth_http_service, + ): + pyramid_request.lti_user = admin_user + oauth_http_service.get.side_effect = self.get_request_handler(is_admin=True) + + url = svc.get_video_download_url("42") + + # When the current user _is_ the admin, we use the normal/default + # OAuthHTTPService instead of the separate admin one. + oauth_http_service.get.assert_called_once() + admin_oauth_http_service.get.assert_not_called() + assert url == "https://videos.cdn.com/video.mp4?signature=abc" + + def test_get_video_download_non_redirect(self, svc): with pytest.raises(ExternalRequestError) as exc_info: svc.get_video_download_url("123") assert exc_info.value.message == "Media download did not return valid redirect" assert Any.instance_of(exc_info.value.response).with_attrs({"status_code": 400}) + def test_get_video_download_error(self, svc): + with pytest.raises(ExternalRequestError) as exc_info: + svc.get_video_download_url("456") + assert Any.instance_of(exc_info.value.response).with_attrs({"status_code": 404}) + + def test_get_video_download_url_fails_if_admin_email_not_set( + self, svc, pyramid_request + ): + pyramid_request.lti_user.application_instance.settings.set( + "canvas_studio", "admin_email", None + ) + with pytest.raises(HTTPBadRequest) as exc_info: + svc.get_video_download_url("42") + assert ( + exc_info.value.message + == "Admin account is not configured for Canvas Studio integration" + ) + + # Test scenario where the admin user never performed an LTI launch, and thus + # we cannot look up the admin email <-> LTI user ID association. + def test_get_video_download_url_fails_if_admin_user_not_in_db( + self, svc, admin_user, db_session + ): + db_session.flush() # Ensure admin user is saved to DB + db_session.delete(admin_user) + + with pytest.raises(HTTPBadRequest) as exc_info: + svc.get_video_download_url("42") + + assert ( + exc_info.value.message + == "The Canvas Studio admin needs to authenticate the Hypothesis integration" + ) + + # Test scenario where the admin user has performed an LTI launch, but has + # not authenticated with Canvas Studio or the token has clearly expired. + def test_get_video_download_url_fails_if_admin_not_authenticated( + self, svc, admin_oauth_http_service + ): + admin_oauth_http_service.get.side_effect = OAuth2TokenError() + + with pytest.raises(HTTPBadRequest) as exc_info: + svc.get_video_download_url("42") + + assert ( + exc_info.value.message + == "The Canvas Studio admin needs to authenticate the Hypothesis integration" + ) + + def test_admin_token_refreshed_if_needed( + self, admin_oauth_http_service, svc, client_secret + ): + # Set up admin-authenticated OAuth request to fail due to expired token. + token_expired_response = factories.requests.Response(status_code=401) + original_get = admin_oauth_http_service.get.side_effect + admin_oauth_http_service.get.side_effect = ExternalRequestError( + response=token_expired_response + ) + + def refresh_ok(*_args, **_kwargs): + admin_oauth_http_service.get.side_effect = original_get + + def refresh_fail(*_args, **_kwargs): + raise ExternalRequestError(message="refresh failed") + + admin_oauth_http_service.refresh_access_token.side_effect = refresh_ok + + # Perform a request that is admin-authenticated. This should trigger + # a refresh and then succeed as normal. + url = svc.get_video_download_url("42") + + admin_oauth_http_service.refresh_access_token.assert_called_with( + "https://hypothesis.instructuremedia.com/api/public/oauth/token", + "http://example.com/api/canvas_studio/oauth/callback", + auth=("the_client_id", client_secret), + ) + assert url == "https://videos.cdn.com/video.mp4?signature=abc" + + # Set up the initial request to fail again, due to an expired token, + # but this time make the refresh fail. + admin_oauth_http_service.get.side_effect = ExternalRequestError( + response=token_expired_response + ) + admin_oauth_http_service.refresh_access_token.side_effect = refresh_fail + + with pytest.raises(ExternalRequestError) as exc_info: + svc.get_video_download_url("42") + + assert ( + exc_info.value.message + == "Canvas Studio admin token refresh failed. Ask the admin user to re-authenticate." + ) + + # Set up the initial request to fail again, due to an expired token, + # but this time make subsequent requests fail even though the refresh + # apparently succeeded. + admin_oauth_http_service.get.side_effect = ExternalRequestError( + response=token_expired_response + ) + admin_oauth_http_service.refresh_access_token.side_effect = None + + with pytest.raises(Exception) as exc_info: + svc.get_video_download_url("42") + def test_get_transcript_url_returns_url_if_published(self, svc): transcript_url = svc.get_transcript_url("42") assert ( @@ -163,8 +290,19 @@ def oauth_http_service(self): svc.get.side_effect = self.get_request_handler() return svc - def get_request_handler(self, collections=None): - """Create a handler for `GET` requests to the Canvas Studio API.""" + @pytest.fixture + def admin_oauth_http_service(self): + svc = create_autospec(OAuthHTTPService, spec_set=True) + svc.get.side_effect = self.get_request_handler(is_admin=True) + return svc + + def get_request_handler(self, collections=None, is_admin=False): + """ + Create a handler for `GET` requests to the Canvas Studio API. + + :param collections: The list of collections the user can access + :param is_admin: Whether to simulate being the admin user + """ def make_collection(id_, name, type_, created_at): return {"id": id_, "name": name, "type": type_, "created_at": created_at} @@ -205,12 +343,14 @@ def handler(url, allow_redirects=True): ] } case "media/42/caption_files": + assert is_admin json_data = { "caption_files": [ {"status": "published", "url": "captions/abc.srt"} ] } case "media/123/caption_files": + assert is_admin json_data = { "caption_files": [ { @@ -219,6 +359,7 @@ def handler(url, allow_redirects=True): ] } case "media/42/download": + assert is_admin assert allow_redirects is False return factories.requests.Response( status_code=302, @@ -227,22 +368,44 @@ def handler(url, allow_redirects=True): }, ) case "media/123/download": - status_code = 400 + # This shouldn't happen, but it simulates what would happen + # if we received a non-4xx/5xx status code that is not a + # redirect. + status_code = 200 + json_data = {} + case "media/456/download": + status_code = 404 json_data = {} case _: # pragma: nocover raise ValueError(f"Unexpected URL {url}") - return factories.requests.Response( - status_code=status_code, json_data=json_data - ) + # This mirrors how HTTPService handles responses based on status code. + response = None + try: + response = factories.requests.Response( + status_code=status_code, json_data=json_data + ) + response.raise_for_status() + return response + except RequestException as err: + raise ExternalRequestError( + request=err.request, response=response + ) from err return handler @pytest.fixture - def oauth_http_factory(self, oauth_http_service, patch): + def oauth_http_factory(self, oauth_http_service, admin_oauth_http_service, patch): factory = patch("lms.services.canvas_studio.oauth_http_factory") - factory.return_value = oauth_http_service + + def create_oauth_http_service(_context, _request, service, user_id=None): + assert service is Service.CANVAS_STUDIO + if user_id == "admin_user_id": + return admin_oauth_http_service + return oauth_http_service + + factory.side_effect = create_oauth_http_service return factory @pytest.fixture @@ -254,6 +417,9 @@ def client_secret(self, pyramid_request, aes_service): @pytest.fixture def canvas_studio_settings(self, pyramid_request, aes_service): application_instance = pyramid_request.lti_user.application_instance + application_instance.settings.set( + "canvas_studio", "admin_email", "admin@hypothesis.edu" + ) application_instance.settings.set("canvas_studio", "client_id", "the_client_id") application_instance.settings.set( "canvas_studio", "domain", "hypothesis.instructuremedia.com" @@ -262,6 +428,16 @@ def canvas_studio_settings(self, pyramid_request, aes_service): aes_service, "canvas_studio", "client_secret", "the_client_secret" ) + @pytest.fixture + def admin_user(self, db_session, pyramid_request): + user = factories.User( + email="admin@hypothesis.edu", + user_id="admin_user_id", + application_instance=pyramid_request.lti_user.application_instance, + ) + db_session.add(user) + return user + class TestFactory: def test_it(self, pyramid_request, CanvasStudioService): diff --git a/tests/unit/lms/services/oauth_http_test.py b/tests/unit/lms/services/oauth_http_test.py index cb18d0edfe..b3c30a8b9c 100644 --- a/tests/unit/lms/services/oauth_http_test.py +++ b/tests/unit/lms/services/oauth_http_test.py @@ -268,10 +268,30 @@ def test_with_custom_service( ) assert service == OAuthHTTPService.return_value + def test_with_custom_user_id( + self, + OAuthHTTPService, + pyramid_request, + http_service, + oauth2_token_service_factory, + ): + service = factory(sentinel.context, pyramid_request, user_id="custom_user_id") + oauth2_token_service_factory.assert_called_once_with( + sentinel.context, pyramid_request, "custom_user_id" + ) + OAuthHTTPService.assert_called_once_with( + http_service, oauth2_token_service_factory.return_value, Service.LMS + ) + assert service == OAuthHTTPService.return_value + @pytest.fixture def OAuthHTTPService(self, patch): return patch("lms.services.oauth_http.OAuthHTTPService") + @pytest.fixture + def oauth2_token_service_factory(self, patch): + return patch("lms.services.oauth_http.oauth2_token_service_factory") + @pytest.fixture(autouse=True) def OAuthTokenResponseSchema(patch):