From dea788ab0866815192cbab93828b7d6c8f501773 Mon Sep 17 00:00:00 2001 From: Greg Werner Date: Tue, 13 Apr 2021 15:36:34 -0400 Subject: [PATCH 1/2] refactor authenticator, validator, and add tests Signed-off-by: Greg Werner --- ltiauthenticator/lti11/auth.py | 93 +++++++----- ltiauthenticator/lti11/constants.py | 70 +++++++++ ltiauthenticator/lti11/validator.py | 161 ++++++++++++-------- tests/conftest.py | 37 +++++ tests/test_lti11_authenticator.py | 226 ++++++++++------------------ 5 files changed, 339 insertions(+), 248 deletions(-) create mode 100644 ltiauthenticator/lti11/constants.py diff --git a/ltiauthenticator/lti11/auth.py b/ltiauthenticator/lti11/auth.py index 6bc9cbb..4e55788 100644 --- a/ltiauthenticator/lti11/auth.py +++ b/ltiauthenticator/lti11/auth.py @@ -1,70 +1,84 @@ -from tornado import gen - -from traitlets import Dict - +from jupyterhub.app import JupyterHub from jupyterhub.auth import Authenticator +from jupyterhub.handlers import BaseHandler from jupyterhub.utils import url_path_join +from traitlets.config import Dict + from ltiauthenticator.lti11.handlers import LTI11AuthenticateHandler from ltiauthenticator.lti11.validator import LTI11LaunchValidator +from ltiauthenticator.utils import convert_request_to_dict +from ltiauthenticator.utils import get_client_protocol class LTI11Authenticator(Authenticator): """ - JupyterHub Authenticator for use with LTI based services (EdX, Canvas, etc) + JupyterHub LTI 1.1 Authenticator which extends the ltiauthenticator.LTIAuthenticator class. + Messages sent to this authenticator are sent from a tool consumer (TC), such as + an LMS. JupyterHub, as the authenticator, works as the tool provider (TP), also + known as the external tool. + + The LTIAuthenticator base class defines the consumers, defined as 1 or (n) consumer key + and shared secret k/v's to verify requests from their tool consumer. """ auto_login = True - login_service = "LTI" + login_service = "LTI 1.1" consumers = Dict( {}, config=True, help=""" A dict of consumer keys mapped to consumer secrets for those keys. - Allows multiple consumers to securely send users to this JupyterHub instance. """, ) - def get_handlers(self, app): + def get_handlers(self, app: JupyterHub) -> BaseHandler: return [("/lti/launch", LTI11AuthenticateHandler)] - @gen.coroutine - def authenticate(self, handler, data) -> dict: - # FIXME: Run a process that cleans up old nonces every other minute + def login_url(self, base_url): + return url_path_join(base_url, "/lti/launch") + + async def authenticate( # noqa: C901 + self, handler: BaseHandler, data: dict = None + ) -> dict: # noqa: C901 + """ + LTI 1.1 Authenticator. One or more consumer keys/values must be set in the jupyterhub config with the + LTI11Authenticator.consumers dict. + + Args: + handler: JupyterHub's Authenticator handler object. For LTI 1.1 requests, the handler is + an instance of LTIAuthenticateHandler. + data: optional data object + + Returns: + Authentication dictionary + + Raises: + HTTPError if the required values are not in the request + """ validator = LTI11LaunchValidator(self.consumers) - args = {} - for k, values in handler.request.body_arguments.items(): - args[k] = ( - values[0].decode() if len(values) == 1 else [v.decode() for v in values] - ) - - # handle multiple layers of proxied protocol (comma separated) and take the outermost - # value (first from the list) - if "x-forwarded-proto" in handler.request.headers: - # x-forwarded-proto might contain comma delimited values - # left-most value is the one sent by original client - hops = [ - h.strip() - for h in handler.request.headers["x-forwarded-proto"].split(",") - ] - protocol = hops[0] - else: - protocol = handler.request.protocol - - launch_url = protocol + "://" + handler.request.host + handler.request.uri + self.log.debug( + "Original arguments received in request: %s" % handler.request.arguments + ) - if validator.validate_launch_request(launch_url, handler.request.headers, args): - # Before we return lti_user_id, check to see if a canvas_custom_user_id was sent. - # If so, this indicates two things: - # 1. The request was sent from Canvas, not edX - # 2. The request was sent from a Canvas course not running in anonymous mode - # If this is the case we want to use the canvas ID to allow grade returns through the Canvas API - # If Canvas is running in anonymous mode, we'll still want the 'user_id' (which is the `lti_user_id``) + # extract the request arguments to a dict + args = convert_request_to_dict(handler.request.arguments) + self.log.debug("Decoded args from request: %s" % args) + + # get the origin protocol + protocol = get_client_protocol(handler) + self.log.debug("Origin protocol is: %s" % protocol) + + # build the full launch url value required for oauth1 signatures + launch_url = f"{protocol}://{handler.request.host}{handler.request.uri}" + self.log.debug("Launch url is: %s" % launch_url) + if validator.validate_launch_request(launch_url, handler.request.headers, args): + # get the lms vendor to implement optional logic for said vendor canvas_id = handler.get_body_argument("custom_canvas_user_id", default=None) if canvas_id is not None: @@ -78,6 +92,3 @@ def authenticate(self, handler, data) -> dict: k: v for k, v in args.items() if not k.startswith("oauth_") }, } - - def login_url(self, base_url): - return url_path_join(base_url, "/lti/launch") diff --git a/ltiauthenticator/lti11/constants.py b/ltiauthenticator/lti11/constants.py new file mode 100644 index 0000000..23344c3 --- /dev/null +++ b/ltiauthenticator/lti11/constants.py @@ -0,0 +1,70 @@ +# LTI 1.1 +# Defined from https://www.imsglobal.org/specs/ltiv1p1p1/implementation-guide +# We define the user_id as required even though it is defined as recommended as it used as +# a fallback id for nbgrader's lms_user_id column. +LTI11_LAUNCH_PARAMS_REQUIRED = [ + "lti_message_type", + "lti_version", + "resource_link_id", + "user_id", +] + +LTI11_LAUNCH_PARAMS_RECOMMENDED = [ + "resource_link_title", + "roles", + "lis_person_name_given", + "lis_person_name_family", + "lis_person_name_full", + "lis_person_contact_email_primary", + "context_id", + "context_title", + "context_label", + "launch_presentation_locale", + "launch_presentation_document_target", + "launch_presentation_width", + "launch_presentation_height", + "launch_presentation_return_url", + "tool_consumer_info_product_family_code", + "tool_consumer_info_version", + "tool_consumer_instance_guid", + "tool_consumer_instance_name", + "tool_consumer_instance_contact_email", +] + +LTI11_LAUNCH_PARAMS_OTIONAL = [ + "resource_link_description", + "user_image", + "role_scope_mentor", + "context_type", + "launch_presentation_css_url", + "tool_consumer_instance_description", + "tool_consumer_instance_url", +] + +LTI11_LIS_OPTION = [ + "lis_outcome_service_url", + "lis_result_sourcedid", + "lis_person_sourcedid", + "lis_course_offering_sourcedid", + "lis_course_section_sourcedid", +] + +# https://www.imsglobal.org/specs/ltiv1p1/implementation-guide +# Section 4.2 +LTI11_OAUTH_ARGS = [ + "oauth_consumer_key", + "oauth_signature_method", + "oauth_timestamp", + "oauth_nonce", + "oauth_callback", + "oauth_version", + "oauth_signature", +] + +LTI11_LAUNCH_PARAMS_REQUIRED = LTI11_LAUNCH_PARAMS_REQUIRED + LTI11_OAUTH_ARGS + +LTI11_LAUNCH_PARAMS_ALL = ( + LTI11_LAUNCH_PARAMS_REQUIRED + + LTI11_LAUNCH_PARAMS_RECOMMENDED + + LTI11_LAUNCH_PARAMS_OTIONAL +) diff --git a/ltiauthenticator/lti11/validator.py b/ltiauthenticator/lti11/validator.py index 52a4dfc..db10f12 100644 --- a/ltiauthenticator/lti11/validator.py +++ b/ltiauthenticator/lti11/validator.py @@ -1,15 +1,35 @@ +from collections import OrderedDict import time -from tornado import web - from oauthlib.oauth1.rfc5849 import signature -from collections import OrderedDict +from tornado.web import HTTPError + +from traitlets.config import LoggingConfigurable + +from typing import Any +from typing import Dict + +from .constants import LTI11_OAUTH_ARGS +from .constants import LTI11_LAUNCH_PARAMS_REQUIRED + + +class LTI11LaunchValidator(LoggingConfigurable): + """ + This class closely mimics the jupyterhub/ltiauthenticator LTILaunchValidator + base class. Inherits from the LoggingConfigurable traitlet to support logging. + + Allows JupyterHub to verify LTI 1.1 compatible requests as a tool + provider (TP). + + For an instance of this class to work, you need to set the consumer key and + shared secret key(s)/value(s) in `LTI11Authenticator` settings, which inherits + from the ``ltiauthenticator.LTIAuthenticator`` class. The key/value pairs are + set as are defined as a dict using the ``consumers`` attribute. -class LTI11LaunchValidator: - # Record time when process starts, so we can reject requests made - # before this - PROCESS_START_TIME = int(time.time()) + Attributes: + consumers: consumer key and shared secret key/value pair(s) + """ # Keep a class-wide, global list of nonces so we can detect & reject # replay attacks. This possibly makes this non-threadsafe, however. @@ -18,57 +38,81 @@ class LTI11LaunchValidator: def __init__(self, consumers): self.consumers = consumers - def validate_launch_request(self, launch_url, headers, args): - """ - Validate a given launch request - - launch_url: Full URL that the launch request was POSTed to - headers: k/v pair of HTTP headers coming in with the POST - args: dictionary of body arguments passed to the launch_url - Must have the following keys to be valid: - oauth_consumer_key, oauth_timestamp, oauth_nonce, - oauth_signature + def validate_launch_request( + self, + launch_url: str, + headers: Dict[str, Any], + args: Dict[str, Any], + ) -> bool: """ + Validate a given LTI 1.1 launch request. The arguments' k/v's are either + required, recommended, or optional. The required/recommended/optional + keys are defined as constants. - # Validate args! - if "oauth_consumer_key" not in args: - raise web.HTTPError(401, "oauth_consumer_key missing") - if args["oauth_consumer_key"] not in self.consumers: - raise web.HTTPError(401, "oauth_consumer_key not known") - - if "oauth_signature" not in args: - raise web.HTTPError(401, "oauth_signature missing") - if "oauth_timestamp" not in args: - raise web.HTTPError(401, "oauth_timestamp missing") - - # Allow 30s clock skew between LTI Consumer and Provider - # Also don't accept timestamps from before our process started, since that could be - # a replay attack - we won't have nonce lists from back then. This would allow users - # who can control / know when our process restarts to trivially do replay attacks. - oauth_timestamp = int(float(args["oauth_timestamp"])) - if ( - int(time.time()) - oauth_timestamp > 30 - or oauth_timestamp < LTI11LaunchValidator.PROCESS_START_TIME - ): - raise web.HTTPError(401, "oauth_timestamp too old") - - if "oauth_nonce" not in args: - raise web.HTTPError(401, "oauth_nonce missing") - if ( - oauth_timestamp in LTI11LaunchValidator.nonces - and args["oauth_nonce"] in LTI11LaunchValidator.nonces[oauth_timestamp] - ): - raise web.HTTPError(401, "oauth_nonce + oauth_timestamp already used") - LTI11LaunchValidator.nonces.setdefault(oauth_timestamp, set()).add( - args["oauth_nonce"] - ) + Args: + launch_url: URL (base_url + path) that receives the launch request, + usually from a tool consumer. + headers: HTTP headers included with the POST request + args: the body sent to the launch url. - args_list = [] - for key, values in args.items(): - if type(values) is list: - args_list += [(key, value) for value in values] - else: - args_list.append((key, values)) + Returns: + True if the validation passes, False otherwise. + + Raises: + HTTPError if a required argument is not inclued in the POST request. + """ + # Ensure that required oauth_* body arguments are included in the request + for param in LTI11_OAUTH_ARGS: + if param not in args.keys(): + raise HTTPError( + 400, "Required oauth arg %s not included in request" % param + ) + if not args.get(param): + raise HTTPError( + 400, "Required oauth arg %s does not have a value" % param + ) + + # Ensure that consumer key is registered in in jupyterhub_config.py + # LTI11Authenticator.consumers defined in parent class + if args["oauth_consumer_key"] not in self.consumers: + raise HTTPError(401, "unknown oauth_consumer_key") + + # Ensure that required LTI 1.1 body arguments are included in the request + for param in LTI11_LAUNCH_PARAMS_REQUIRED: + if param not in args.keys(): + raise HTTPError( + 400, "Required LTI 1.1 arg arg %s not included in request" % param + ) + if not args.get(param): + raise HTTPError( + 400, "Required LTI 1.1 arg %s does not have a value" % param + ) + + # Inspiration to validate nonces/timestamps from OAuthlib + # https://github.com/oauthlib/oauthlib/blob/master/oauthlib/oauth1/rfc5849/endpoints/base.py#L147 + if len(str(int(args["oauth_timestamp"]))) != 10: + raise HTTPError(401, "Invalid timestamp format.") + try: + ts = int(args["oauth_timestamp"]) + except ValueError: + raise HTTPError(401, "Timestamp must be an integer.") + else: + # Reject timestamps that are older than 30 seconds + if abs(time.time() - ts) > 30: + raise HTTPError( + 401, + "Timestamp given is invalid, differ from " + "allowed by over %s seconds." % str(int(time.time() - ts)), + ) + if ( + ts in LTI11LaunchValidator.nonces + and args["oauth_nonce"] in LTI11LaunchValidator.nonces[ts] + ): + raise HTTPError(401, "oauth_nonce + oauth_timestamp already used") + LTI11LaunchValidator.nonces.setdefault(ts, set()).add(args["oauth_nonce"]) + + # convert arguments dict back to a list of tuples for signature + args_list = [(k, v) for k, v in args.items()] base_string = signature.signature_base_string( "POST", @@ -77,13 +121,12 @@ def validate_launch_request(self, launch_url, headers, args): signature.collect_parameters(body=args_list, headers=headers) ), ) - consumer_secret = self.consumers[args["oauth_consumer_key"]] - sign = signature.sign_hmac_sha1(base_string, consumer_secret, None) is_valid = signature.safe_string_equals(sign, args["oauth_signature"]) - + self.log.debug("signature in request: %s" % args["oauth_signature"]) + self.log.debug("calculated signature: %s" % sign) if not is_valid: - raise web.HTTPError(401, "Invalid oauth_signature") + raise HTTPError(401, "Invalid oauth_signature") return True diff --git a/tests/conftest.py b/tests/conftest.py index 250ce6b..d19e01a 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -385,3 +385,40 @@ def _make_lti11_success_authentication_request_args( return args return _make_lti11_success_authentication_request_args + + +@pytest.fixture(scope="function") +def make_mock_request_handler() -> RequestHandler: + """ + Sourced from https://github.com/jupyterhub/oauthenticator/blob/master/oauthenticator/tests/mocks.py + """ + + def _make_mock_request_handler( + handler: RequestHandler, + uri: str = "https://hub.example.com", + method: str = "GET", + **settings: dict, + ) -> RequestHandler: + """Instantiate a Handler in a mock application""" + application = Application( + hub=Mock( + base_url="/hub/", + server=Mock(base_url="/hub/"), + ), + cookie_secret=os.urandom(32), + db=Mock(rollback=Mock(return_value=None)), + **settings, + ) + request = HTTPServerRequest( + method=method, + uri=uri, + connection=Mock(), + ) + handler = RequestHandler( + application=application, + request=request, + ) + handler._transforms = [] + return handler + + return _make_mock_request_handler diff --git a/tests/test_lti11_authenticator.py b/tests/test_lti11_authenticator.py index acaf79b..0b62ca9 100644 --- a/tests/test_lti11_authenticator.py +++ b/tests/test_lti11_authenticator.py @@ -1,130 +1,77 @@ import pytest -import time - -from oauthlib.oauth1.rfc5849 import signature - from tornado import web from ltiauthenticator.lti11.validator import LTI11LaunchValidator -def make_args( - consumer_key, consumer_secret, launch_url, oauth_timestamp, oauth_nonce, extra_args -): - args = { - "oauth_consumer_key": consumer_key, - "oauth_timestamp": str(oauth_timestamp), - "oauth_nonce": oauth_nonce, - } - - args.update(extra_args) - - base_string = signature.signature_base_string( - "POST", - signature.base_string_uri(launch_url), - signature.normalize_parameters( - signature.collect_parameters(body=args, headers={}) - ), - ) - - args["oauth_signature"] = signature.sign_hmac_sha1( - base_string, consumer_secret, None - ) +def test_launch(make_lti11_basic_launch_request_args): + """Test a basic launch request""" + oauth_consumer_key = "my_consumer_key" + oauth_consumer_secret = "my_shared_secret" + launch_url = "http://jupyterhub/hub/lti/launch" + headers = {"Content-Type": "application/x-www-form-urlencoded"} - return args - - -def test_launch(): - consumer_key = "key1" - consumer_secret = "secret1" - launch_url = "http://localhost:8000/hub/lti/launch" - headers = {} - oauth_timestamp = time.time() - oauth_nonce = str(time.time()) - extra_args = {"arg1": "value1"} - - args = make_args( - consumer_key, - consumer_secret, - launch_url, - oauth_timestamp, - oauth_nonce, - extra_args, + args = make_lti11_basic_launch_request_args( + oauth_consumer_key, + oauth_consumer_secret, ) - validator = LTI11LaunchValidator({consumer_key: consumer_secret}) + validator = LTI11LaunchValidator({oauth_consumer_key: oauth_consumer_secret}) assert validator.validate_launch_request(launch_url, headers, args) -def test_wrong_key(): - consumer_key = "key1" - consumer_secret = "secret1" - launch_url = "http://localhost:8000/hub/lti/launch" - headers = {} - oauth_timestamp = time.time() - oauth_nonce = str(time.time()) - extra_args = {"arg1": "value1"} - - args = make_args( - consumer_key, - consumer_secret, - launch_url, - oauth_timestamp, - oauth_nonce, - extra_args, +def test_wrong_key(make_lti11_basic_launch_request_args): + """Test that the request is rejected when receiving the wrong consumer key.""" + oauth_consumer_key = "my_consumer_key" + oauth_consumer_secret = "my_shared_secret" + launch_url = "http://jupyterhub/hub/lti/launch" + headers = {"Content-Type": "application/x-www-form-urlencoded"} + + args = make_lti11_basic_launch_request_args( + oauth_consumer_key, + oauth_consumer_secret, ) - validator = LTI11LaunchValidator({"wrongkey": consumer_secret}) + validator = LTI11LaunchValidator({"wrongkey": oauth_consumer_secret}) with pytest.raises(web.HTTPError): assert validator.validate_launch_request(launch_url, headers, args) -def test_wrong_secret(): - consumer_key = "key1" - consumer_secret = "secret1" - launch_url = "http://localhost:8000/hub/lti/launch" - headers = {} - oauth_timestamp = time.time() - oauth_nonce = str(time.time()) - extra_args = {"arg1": "value1"} - - args = make_args( - consumer_key, - consumer_secret, - launch_url, - oauth_timestamp, - oauth_nonce, - extra_args, +def test_wrong_secret(make_lti11_basic_launch_request_args): + """Test that a request is rejected when the signature is created with the wrong secret.""" + oauth_consumer_key = "my_consumer_key" + oauth_consumer_secret = "my_shared_secret" + launch_url = "http://jupyterhub/hub/lti/launch" + headers = {"Content-Type": "application/x-www-form-urlencoded"} + + args = make_lti11_basic_launch_request_args( + oauth_consumer_key, + oauth_consumer_secret, ) - validator = LTI11LaunchValidator({consumer_key: "wrongsecret"}) + validator = LTI11LaunchValidator({oauth_consumer_key: "wrongsecret"}) with pytest.raises(web.HTTPError): validator.validate_launch_request(launch_url, headers, args) -def test_full_replay(): - consumer_key = "key1" - consumer_secret = "secret1" - launch_url = "http://localhost:8000/hub/lti/launch" - headers = {} - oauth_timestamp = time.time() - oauth_nonce = str(time.time()) - extra_args = {"arg1": "value1"} - - args = make_args( - consumer_key, - consumer_secret, - launch_url, - oauth_timestamp, - oauth_nonce, - extra_args, +def test_full_replay(make_lti11_basic_launch_request_args): + """Ensure that an oauth timestamp/nonce replay raises an HTTPError""" + + oauth_consumer_key = "my_consumer_key" + oauth_consumer_secret = "my_shared_secret" + launch_url = "http://jupyterhub/hub/lti/launch" + headers = {"Content-Type": "application/x-www-form-urlencoded"} + + args = make_lti11_basic_launch_request_args( + oauth_consumer_key, + oauth_consumer_secret, ) - validator = LTI11LaunchValidator({consumer_key: consumer_secret}) + validator = LTI11LaunchValidator({oauth_consumer_key: oauth_consumer_secret}) assert validator.validate_launch_request(launch_url, headers, args) @@ -132,25 +79,20 @@ def test_full_replay(): validator.validate_launch_request(launch_url, headers, args) -def test_partial_replay_timestamp(): - consumer_key = "key1" - consumer_secret = "secret1" - launch_url = "http://localhost:8000/hub/lti/launch" - headers = {} - oauth_timestamp = time.time() - oauth_nonce = str(time.time()) - extra_args = {"arg1": "value1"} - - args = make_args( - consumer_key, - consumer_secret, - launch_url, - oauth_timestamp, - oauth_nonce, - extra_args, +def test_partial_replay_timestamp(make_lti11_basic_launch_request_args): + """Test that a partial timestamp replay raises an HTTPError.""" + + oauth_consumer_key = "my_consumer_key" + oauth_consumer_secret = "my_shared_secret" + launch_url = "http://jupyterhub/hub/lti/launch" + headers = {"Content-Type": "application/x-www-form-urlencoded"} + + args = make_lti11_basic_launch_request_args( + oauth_consumer_key, + oauth_consumer_secret, ) - validator = LTI11LaunchValidator({consumer_key: consumer_secret}) + validator = LTI11LaunchValidator({oauth_consumer_key: oauth_consumer_secret}) assert validator.validate_launch_request(launch_url, headers, args) @@ -159,25 +101,19 @@ def test_partial_replay_timestamp(): validator.validate_launch_request(launch_url, headers, args) -def test_partial_replay_nonce(): - consumer_key = "key1" - consumer_secret = "secret1" - launch_url = "http://localhost:8000/hub/lti/launch" - headers = {} - oauth_timestamp = time.time() - oauth_nonce = str(time.time()) - extra_args = {"arg1": "value1"} - - args = make_args( - consumer_key, - consumer_secret, - launch_url, - oauth_timestamp, - oauth_nonce, - extra_args, +def test_partial_replay_nonce(make_lti11_basic_launch_request_args): + """Test that a partial nonce replay raises an HTTPError""" + oauth_consumer_key = "my_consumer_key" + oauth_consumer_secret = "my_shared_secret" + launch_url = "http://jupyterhub/hub/lti/launch" + headers = {"Content-Type": "application/x-www-form-urlencoded"} + + args = make_lti11_basic_launch_request_args( + oauth_consumer_key, + oauth_consumer_secret, ) - validator = LTI11LaunchValidator({consumer_key: consumer_secret}) + validator = LTI11LaunchValidator({oauth_consumer_key: oauth_consumer_secret}) assert validator.validate_launch_request(launch_url, headers, args) @@ -186,25 +122,19 @@ def test_partial_replay_nonce(): validator.validate_launch_request(launch_url, headers, args) -def test_dubious_extra_args(): - consumer_key = "key1" - consumer_secret = "secret1" - launch_url = "http://localhost:8000/hub/lti/launch" - headers = {} - oauth_timestamp = time.time() - oauth_nonce = str(time.time()) - extra_args = {"arg1": "value1"} - - args = make_args( - consumer_key, - consumer_secret, - launch_url, - oauth_timestamp, - oauth_nonce, - extra_args, +def test_dubious_extra_args(make_lti11_basic_launch_request_args): + """Ensure that dubious extra args are rejected""" + oauth_consumer_key = "my_consumer_key" + oauth_consumer_secret = "my_shared_secret" + launch_url = "http://jupyterhub/hub/lti/launch" + headers = {"Content-Type": "application/x-www-form-urlencoded"} + + args = make_lti11_basic_launch_request_args( + oauth_consumer_key, + oauth_consumer_secret, ) - validator = LTI11LaunchValidator({consumer_key: consumer_secret}) + validator = LTI11LaunchValidator({oauth_consumer_key: oauth_consumer_secret}) assert validator.validate_launch_request(launch_url, headers, args) From 1a52ee6f66d0c741951b0b7b8c0148e6f77d5fea Mon Sep 17 00:00:00 2001 From: Greg Werner Date: Wed, 14 Apr 2021 14:33:39 -0400 Subject: [PATCH 2/2] remove unused fixtures Signed-off-by: Greg Werner --- tests/conftest.py | 374 ---------------------------------------------- 1 file changed, 374 deletions(-) diff --git a/tests/conftest.py b/tests/conftest.py index d19e01a..cbe8665 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,284 +1,12 @@ -import json -import os import secrets import time -import uuid -from io import StringIO from oauthlib.oauth1.rfc5849 import signature import pytest -from tornado.httpclient import AsyncHTTPClient -from tornado.httpclient import HTTPResponse -from tornado.httputil import HTTPHeaders -from tornado.httputil import HTTPServerRequest -from tornado.web import Application -from tornado.web import RequestHandler - from typing import Dict -from unittest.mock import Mock -from unittest.mock import patch - - -@pytest.fixture(scope="module") -def auth_state_dict(): - authenticator_auth_state = { - "name": "student1", - "auth_state": { - "course_id": "intro101", - "course_lineitems": "my.platform.com/api/lti/courses/1/line_items", - "lms_user_id": "185d6c59731a553009ca9b59ca3a885100000", - "user_role": "Learner", - }, - } - return authenticator_auth_state - - -@pytest.fixture(scope="module") -def app(): - class TestHandler(RequestHandler): - def get(self): - self.write("test") - - def post(self): - self.write("test") - - application = Application( - [ - (r"/", TestHandler), - ] - ) # noqa: E231 - return application - - -@pytest.fixture(scope="function") -def jupyterhub_api_environ(monkeypatch): - """ - Set the enviroment variables used in Course class - """ - monkeypatch.setenv("JUPYTERHUB_API_TOKEN", str(uuid.uuid4())) - monkeypatch.setenv("JUPYTERHUB_API_URL", "https://localhost/hub/api") - monkeypatch.setenv("JUPYTERHUB_ADMIN_USER", "admin") - - -@pytest.fixture(scope="function") -def lti11_config_environ(monkeypatch, pem_file): - """ - Set the enviroment variables used in Course class - """ - monkeypatch.setenv("LTI_CONSUMER_KEY", "ild_test_consumer_key") - monkeypatch.setenv("LTI_SHARED_SECRET", "ild_test_shared_secret") - - -@pytest.fixture(scope="function") -def lti11_complete_launch_args(): - """ - Valid response when retrieving jwks from the platform. - """ - args = { - "oauth_callback": ["about:blank".encode()], - "oauth_consumer_key": ["my_consumer_key".encode()], - "oauth_signature_method": ["HMAC-SHA1".encode()], - "oauth_timestamp": ["1585947271".encode()], - "oauth_nonce": ["01fy8HKIASKuD9gK9vWUcBj9fql1nOCWfOLPzeylsmg".encode()], - "oauth_signature": ["abc123".encode()], - "oauth_version": ["1.0".encode()], - "context_id": ["888efe72d4bbbdf90619353bb8ab5965ccbe9b3f".encode()], - "context_label": ["intro101".encode()], - "context_title": ["intro101".encode()], - "custom_canvas_assignment_title": ["test-assignment".encode()], - "custom_canvas_user_login_id": ["student1".encode()], - "custom_worskpace_type": ["foo".encode()], - "ext_roles": ["urn:lti:instrole:ims/lis/Learner".encode()], - "launch_presentation_document_target": ["iframe".encode()], - "launch_presentation_height": ["1000".encode()], - "launch_presentation_locale": ["en".encode()], - "launch_presentation_return_url": [ - "https: //illumidesk.instructure.com/courses/161/external_content/success/external_tool_redirect".encode() - ], - "launch_presentation_width": ["1000".encode()], - "lis_outcome_service_url": [ - "http://www.imsglobal.org/developers/LTI/test/v1p1/common/tool_consumer_outcome.php?b64=MTIzNDU6OjpzZWNyZXQ=".encode() - ], - "lis_person_contact_email_primary": ["student1@example.com".encode()], - "lis_person_name_family": ["Bar".encode()], - "lis_person_name_full": ["Foo Bar".encode()], - "lis_person_name_given": ["Foo".encode()], - "lti_message_type": ["basic-lti-launch-request".encode()], - "lis_result_sourcedid": ["feb-123-456-2929::28883".encode()], - "lti_version": ["LTI-1p0".encode()], - "resource_link_id": ["888efe72d4bbbdf90619353bb8ab5965ccbe9b3f".encode()], - "resource_link_title": ["Test-Assignment-Another-LMS".encode()], - "roles": ["Learner".encode()], - "tool_consumer_info_product_family_code": ["canvas".encode()], - "tool_consumer_info_version": ["cloud".encode()], - "tool_consumer_instance_contact_email": ["notifications@mylms.com".encode()], - "tool_consumer_instance_guid": [ - "srnuz6h1U8kOMmETzoqZTJiPWzbPXIYkAUnnAJ4u:test-lms".encode() - ], - "tool_consumer_instance_name": ["myorg".encode()], - "user_id": ["185d6c59731a553009ca9b59ca3a885100000".encode()], - "user_image": ["https://lms.example.com/avatar-50.png".encode()], - } - return args - - -@pytest.fixture -def mock_jhub_user(request): - """ - Creates an Authenticated User mock by returning a wrapper function to help us to customize its creation - Usage: - user_mocked = mock_jhub_user(environ={'USER_ROLE': 'Instructor'}) - or - user_mocked = mock_jhub_user() - or - user_mocked = mock_jhub_user(environ={'USER_ROLE': 'Instructor'}, auth_state=[]) - """ - - def _get_with_params(environ: dict = None, auth_state: list = []) -> Mock: - """ - wrapper function that accept environment and auth_state - Args: - auth_state: Helps with the `the get_auth_state` method - """ - mock_user = Mock() - mock_spawner = Mock() - # define the mock attrs - spawner_attrs = {"environment": environ or {}} - mock_spawner.configure_mock(**spawner_attrs) - attrs = { - "name": "user1", - "spawner": mock_spawner, - "get_auth_state.side_effect": auth_state or [], - } - mock_user.configure_mock(**attrs) - return mock_user - - return _get_with_params - - -@pytest.fixture(scope="function") -def make_mock_request_handler() -> RequestHandler: - """ - Sourced from https://github.com/jupyterhub/oauthenticator/blob/master/oauthenticator/tests/mocks.py - """ - - def _make_mock_request_handler( - handler: RequestHandler, - uri: str = "https://hub.example.com", - method: str = "GET", - **settings: dict, - ) -> RequestHandler: - """Instantiate a Handler in a mock application""" - application = Application( - hub=Mock( - base_url="/hub/", - server=Mock(base_url="/hub/"), - ), - cookie_secret=os.urandom(32), - db=Mock(rollback=Mock(return_value=None)), - **settings, - ) - request = HTTPServerRequest( - method=method, - uri=uri, - connection=Mock(), - ) - handler = RequestHandler( - application=application, - request=request, - ) - handler._transforms = [] - return handler - - return _make_mock_request_handler - - -@pytest.fixture(scope="function") -def make_http_response() -> HTTPResponse: - async def _make_http_response( - handler: RequestHandler, - code: int = 200, - reason: str = "OK", - headers: HTTPHeaders = HTTPHeaders({"content-type": "application/json"}), - effective_url: str = "http://hub.example.com/", - body: Dict[str, str] = {"foo": "bar"}, - ) -> HTTPResponse: - """ - Creates an HTTPResponse object from a given request. - - Args: - handler: tornado.web.RequestHandler object. - code: response code, e.g. 200 or 404 - reason: reason phrase describing the status code - headers: HTTPHeaders (response header object), use the dict within the constructor, e.g. - {"content-type": "application/json"} - effective_url: final location of the resource after following any redirects - body: dictionary that represents the StringIO (buffer) body - - Returns: - A tornado.client.HTTPResponse object - """ - dict_to_buffer = StringIO(json.dumps(body)) if body is not None else None - return HTTPResponse( - request=handler, - code=code, - reason=reason, - headers=headers, - effective_url=effective_url, - buffer=dict_to_buffer, - ) - - return _make_http_response - - -@pytest.fixture(scope="function") -def http_async_httpclient_with_simple_response( - request, make_http_response, make_mock_request_handler -): - """ - Creates a patch of AsyncHttpClient.fetch method, useful when other tests are making http request - """ - local_handler = make_mock_request_handler(RequestHandler) - test_request_body_param = ( - request.param if hasattr(request, "param") else {"message": "ok"} - ) - with patch.object( - AsyncHTTPClient, - "fetch", - return_value=make_http_response( - handler=local_handler.request, body=test_request_body_param - ), - ): - yield AsyncHTTPClient() - - -@pytest.fixture(scope="function") -def make_auth_state_dict() -> Dict[str, str]: - """ - Creates an authentication dictionary with default name and auth_state k/v's - """ - - def _make_auth_state_dict( - username: str = "foo", - assignment_name: str = "myassignment", - course_id: str = "intro101", - lms_user_id: str = "abc123", - user_role: str = "Learner", - ): - return { - "name": username, - "auth_state": { - "assignment_name": assignment_name, - "course_id": course_id, - "lms_user_id": lms_user_id, - }, # noqa: E231 - } - - return _make_auth_state_dict - @pytest.fixture(scope="function") def make_lti11_basic_launch_request_args() -> Dict[str, str]: @@ -320,105 +48,3 @@ def _make_lti11_basic_launch_args( return args return _make_lti11_basic_launch_args - - -@pytest.fixture(scope="function") -def make_lti11_success_authentication_request_args(): - def _make_lti11_success_authentication_request_args( - lms_vendor: str = "canvas", role: str = "Instructor" - ) -> Dict[str, str]: - """ - Return a valid request arguments make from LMS to our tool (when authentication steps were success) - """ - args = { - "oauth_callback": ["about:blank".encode()], - "oauth_consumer_key": ["my_consumer_key".encode()], - "oauth_signature_method": ["HMAC-SHA1".encode()], - "oauth_timestamp": ["1585947271".encode()], - "oauth_nonce": ["01fy8HKIASKuD9gK9vWUcBj9fql1nOCWfOLPzeylsmg".encode()], - "oauth_signature": ["abc123".encode()], - "oauth_version": ["1.0".encode()], - "context_id": ["888efe72d4bbbdf90619353bb8ab5965ccbe9b3f".encode()], - "context_label": ["intro101".encode()], - "context_title": ["intro101".encode()], - "course_lineitems": [ - "my.platform.com/api/lti/courses/1/line_items".encode() - ], - "custom_canvas_assignment_title": ["test-assignment".encode()], - "custom_canvas_course_id": ["616".encode()], - "custom_canvas_enrollment_state": ["active".encode()], - "custom_canvas_user_id": ["1091".encode()], - "custom_canvas_user_login_id": ["student1@example.com".encode()], - "ext_roles": ["urn:lti:instrole:ims/lis/Learner".encode()], - "launch_presentation_document_target": ["iframe".encode()], - "launch_presentation_height": ["1000".encode()], - "launch_presentation_locale": ["en".encode()], - "launch_presentation_return_url": [ - "https: //illumidesk.instructure.com/courses/161/external_content/success/external_tool_redirect".encode() - ], - "launch_presentation_width": ["1000".encode()], - "lis_outcome_service_url": [ - "http://www.imsglobal.org/developers/LTI/test/v1p1/common/tool_consumer_outcome.php?b64=MTIzNDU6OjpzZWNyZXQ=".encode() - ], - "lis_person_contact_email_primary": ["student1@example.com".encode()], - "lis_person_name_family": ["Bar".encode()], - "lis_person_name_full": ["Foo Bar".encode()], - "lis_person_name_given": ["Foo".encode()], - "lti_message_type": ["basic-lti-launch-request".encode()], - "lis_result_sourcedid": ["feb-123-456-2929::28883".encode()], - "lti_version": ["LTI-1p0".encode()], - "resource_link_id": ["888efe72d4bbbdf90619353bb8ab5965ccbe9b3f".encode()], - "resource_link_title": ["Test-Assignment-Another-LMS".encode()], - "roles": [role.encode()], - "tool_consumer_info_product_family_code": [lms_vendor.encode()], - "tool_consumer_info_version": ["cloud".encode()], - "tool_consumer_instance_contact_email": [ - "notifications@mylms.com".encode() - ], - "tool_consumer_instance_guid": [ - "srnuz6h1U8kOMmETzoqZTJiPWzbPXIYkAUnnAJ4u:test-lms".encode() - ], - "tool_consumer_instance_name": ["myorg".encode()], - "user_id": ["185d6c59731a553009ca9b59ca3a885100000".encode()], - "user_image": ["https://lms.example.com/avatar-50.png".encode()], - } - return args - - return _make_lti11_success_authentication_request_args - - -@pytest.fixture(scope="function") -def make_mock_request_handler() -> RequestHandler: - """ - Sourced from https://github.com/jupyterhub/oauthenticator/blob/master/oauthenticator/tests/mocks.py - """ - - def _make_mock_request_handler( - handler: RequestHandler, - uri: str = "https://hub.example.com", - method: str = "GET", - **settings: dict, - ) -> RequestHandler: - """Instantiate a Handler in a mock application""" - application = Application( - hub=Mock( - base_url="/hub/", - server=Mock(base_url="/hub/"), - ), - cookie_secret=os.urandom(32), - db=Mock(rollback=Mock(return_value=None)), - **settings, - ) - request = HTTPServerRequest( - method=method, - uri=uri, - connection=Mock(), - ) - handler = RequestHandler( - application=application, - request=request, - ) - handler._transforms = [] - return handler - - return _make_mock_request_handler