<commit_before>from traitlets import List, Unicode
from oauthenticator.generic import GenericOAuthenticator
import aiohttp

class CanvasOAuthenticator(GenericOAuthenticator):
    """
    Canvas OAuth2 based authenticator for JupyterHub.

    Collects info about user & enrolled courses from canvas,
    puts them into auth_state. To refresh, user has to re-login.
    """

    strip_email_domain = Unicode(
        '',
        config=True,
        help="""
        Strip this domain from user emails when making their JupyterHub user name.

        For example, if almost all your users have emails of form <EMAIL>,
        you can set this to 'berkeley.edu'. A canvas user with email <EMAIL>
        will get a JupyterHub user name of 'yuvipanda', while a canvas user with email
        <EMAIL> will get a JupyterHub username of '<EMAIL>'.

        By default, *no* domain stripping is performed, and the JupyterHub username
        is the primary email of the canvas user.
        """
    )

    canvas_url = Unicode(
        '',
        config=True,
        help="""
        URL to canvas installation to use for authentication.

        Must have a trailing slash
        """
    )


    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)

        if not self.canvas_url:
            raise ValueError('c.CanvasOAuthenticator.canvas_url must be set')

        # canvas_url must have a trailing slash
        if self.canvas_url[-1] != '/':
            raise ValueError('c.CanvasOAuthenticator.canvas_url must have a trailing slash')

        self.token_url = f'{self.canvas_url}login/oauth2/token'
        self.userdata_url = f'{self.canvas_url}api/v1/users/self/profile'

        self.extra_params = {
            'client_id': self.client_id,
            'client_secret': self.client_secret,
            'replace_tokens': 'true', # to prevent tokens from accumulating
        }

    async def get_canvas_items(self, token, url):
        """
        Get paginated items from Canvas.
        https://canvas.instructure.com/doc/api/file.pagination.html
        """
        headers = dict(Authorization = f"Bearer {token}")
        data = []

        async with aiohttp.ClientSession() as session:
            async with session.get(url, headers=headers, params=self.extra_params) as r:
                if r.status != 200:
                    raise Exception(f"error fetching items for {self.username}: {url} -- {r.status} -- {r.text()}")
                data = await r.json()
                if 'next' in r.links.keys():
                    url = r.links['next']['url']
                    data += await self.get_canvas_items(token, url)

        return data

    async def get_courses(self, token):
        """
        Get list of courses enrolled by the current user
        """
        url = f"{self.canvas_url}/api/v1/courses"

        data = await self.get_canvas_items(token, url)

        return data

    async def authenticate(self, handler, data=None):
        """
        Augment base user auth info with course info
        """
        user = await super().authenticate(handler, data)
        user['auth_state']['courses'] = await self.get_courses(
            user['auth_state']['access_token']
        )
        return user

    def normalize_username(self, username):
        username = username.lower()
        # To make life easier & match usernames with existing users who were
        # created with google auth, we want to strip the domain name. If not,
        # we use the full email as the official user name
        if self.strip_email_domain and username.endswith('@' + self.strip_email_domain):
            return username.split('@')[0]
        return username

    async def pre_spawn_start(self, user, spawner):
        """Pass oauth data to spawner via OAUTH2_ prefixed env variables."""
        auth_state = yield user.get_auth_state()
        if not auth_state:
            return
        if 'access_token' in auth_state:
            spawner.environment["OAUTH2_ACCESS_TOKEN"] = auth_state['access_token']
        # others are lti_user_id, id, integration_id
        if 'oauth_user' in auth_state:
            for k in ['login_id', 'name', 'sortable_name', 'primary_email']:
                if k in auth_state['oauth_user']:
                    spawner.environment[f"OAUTH2_{k.upper()}"] = auth_state['oauth_user'][k]
<commit_msg>Use 1 for replace_tokens.<commit_after>from traitlets import List, Unicode
from oauthenticator.generic import GenericOAuthenticator
import aiohttp

class CanvasOAuthenticator(GenericOAuthenticator):
    """
    Canvas OAuth2 based authenticator for JupyterHub.

    Collects info about user & enrolled courses from canvas,
    puts them into auth_state. To refresh, user has to re-login.
    """

    strip_email_domain = Unicode(
        '',
        config=True,
        help="""
        Strip this domain from user emails when making their JupyterHub user name.

        For example, if almost all your users have emails of form <EMAIL>,
        you can set this to 'berkeley.edu'. A canvas user with email <EMAIL>
        will get a JupyterHub user name of 'yuvipanda', while a canvas user with email
        <EMAIL> will get a JupyterHub username of '<EMAIL>'.

        By default, *no* domain stripping is performed, and the JupyterHub username
        is the primary email of the canvas user.
        """
    )

    canvas_url = Unicode(
        '',
        config=True,
        help="""
        URL to canvas installation to use for authentication.

        Must have a trailing slash
        """
    )


    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)

        if not self.canvas_url:
            raise ValueError('c.CanvasOAuthenticator.canvas_url must be set')

        # canvas_url must have a trailing slash
        if self.canvas_url[-1] != '/':
            raise ValueError('c.CanvasOAuthenticator.canvas_url must have a trailing slash')

        self.token_url = f'{self.canvas_url}login/oauth2/token'
        self.userdata_url = f'{self.canvas_url}api/v1/users/self/profile'

        self.extra_params = {
            'client_id': self.client_id,
            'client_secret': self.client_secret,
            'replace_tokens': 1, # to prevent tokens from accumulating
        }

    async def get_canvas_items(self, token, url):
        """
        Get paginated items from Canvas.
        https://canvas.instructure.com/doc/api/file.pagination.html
        """
        headers = dict(Authorization = f"Bearer {token}")
        data = []

        async with aiohttp.ClientSession() as session:
            async with session.get(url, headers=headers, params=self.extra_params) as r:
                if r.status != 200:
                    raise Exception(f"error fetching items for {self.username}: {url} -- {r.status} -- {r.text()}")
                data = await r.json()
                if 'next' in r.links.keys():
                    url = r.links['next']['url']
                    data += await self.get_canvas_items(token, url)

        return data

    async def get_courses(self, token):
        """
        Get list of courses enrolled by the current user
        """
        url = f"{self.canvas_url}/api/v1/courses"

        data = await self.get_canvas_items(token, url)

        return data

    async def authenticate(self, handler, data=None):
        """
        Augment base user auth info with course info
        """
        user = await super().authenticate(handler, data)
        user['auth_state']['courses'] = await self.get_courses(
            user['auth_state']['access_token']
        )
        return user

    def normalize_username(self, username):
        username = username.lower()
        # To make life easier & match usernames with existing users who were
        # created with google auth, we want to strip the domain name. If not,
        # we use the full email as the official user name
        if self.strip_email_domain and username.endswith('@' + self.strip_email_domain):
            return username.split('@')[0]
        return username

    async def pre_spawn_start(self, user, spawner):
        """Pass oauth data to spawner via OAUTH2_ prefixed env variables."""
        auth_state = yield user.get_auth_state()
        if not auth_state:
            return
        if 'access_token' in auth_state:
            spawner.environment["OAUTH2_ACCESS_TOKEN"] = auth_state['access_token']
        # others are lti_user_id, id, integration_id
        if 'oauth_user' in auth_state:
            for k in ['login_id', 'name', 'sortable_name', 'primary_email']:
                if k in auth_state['oauth_user']:
                    spawner.environment[f"OAUTH2_{k.upper()}"] = auth_state['oauth_user'][k]

{'fraction_non_alphanumeric': 0.09732498577120091, 'fraction_numerical': 0.0029595902105862266, 'mean_word_length': 2.304249717939075, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 11, 'https://': 2, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2495439', 'n_tokens_mistral': 2521, 'n_tokens_neox': 2448, 'n_words': 757}