lms.backend.moodle.backend

  1# pylint: disable=abstract-method
  2
  3import json
  4import logging
  5import re
  6import typing
  7import urllib.parse
  8
  9import bs4
 10import edq.net.request
 11import requests
 12
 13import lms.backend.moodle.errors
 14import lms.model.backend
 15import lms.model.constants
 16import lms.util.net
 17
 18_logger = logging.getLogger(__name__)
 19
 20ROLE_MAPPING: typing.Dict[str, lms.model.users.CourseRole] = {
 21    "guest": lms.model.users.CourseRole.OTHER,
 22    "student": lms.model.users.CourseRole.STUDENT,
 23    "non-editing teacher": lms.model.users.CourseRole.GRADER,
 24    "teacher": lms.model.users.CourseRole.ADMIN,
 25    "manager": lms.model.users.CourseRole.OWNER,
 26}
 27
 28# Moodle shows 5000 users per page when asked to fetch all results.
 29RESULTS_PER_PAGE: int = 5000
 30
 31class MoodleBackend(lms.model.backend.APIBackend):
 32    """ An API backend for the Moodle LMS. """
 33
 34    def __init__(self,
 35            **kwargs: typing.Any) -> None:
 36        super().__init__(**kwargs)
 37
 38        assert(self.config.backend_type == lms.model.constants.BackendType.MOODLE)
 39
 40        if (self.config.auth_user is None):
 41            raise ValueError("Moodle backends require a username.")
 42
 43        self.auth_user: str = self.config.auth_user
 44        """
 45        The user to authenticate with.
 46        This is set in config and compied for type checking.
 47        """
 48
 49        if (self.config.auth_password is None):
 50            raise ValueError("Moodle backends require a password.")
 51
 52        self.auth_password: str = self.config.auth_password.cleartext
 53        """
 54        The (cleartext) password to authenticate with.
 55        This is set in config and compied for type checking.
 56        """
 57
 58        self._session_headers: typing.Union[typing.Dict[str, typing.Any], None] = None
 59        """ The headers (e.g., cookies) for our logged in Moodle session. """
 60
 61    def _get_edit_mode_page(self, url: str, **kwargs: typing.Any) -> typing.Union[requests.Response, None]:
 62        """
 63        Tries to fetch the page at the given url with edit mode enabled.
 64        Returns the response with edit mode enabled, or None if unable to get the edit mode version of the page.
 65        """
 66
 67        try:
 68            response, _ = edq.net.request.make_get(url, headers = self.get_standard_headers(), **kwargs)
 69        except requests.exceptions.HTTPError:
 70            return None
 71
 72        sesskey_match = re.search(r'"sesskey":"([^"]+)"', response.text)
 73        if (sesskey_match is None):
 74            raise lms.backend.moodle.errors.MoodleAPIBreakageError()
 75
 76        sesskey = sesskey_match.group(1)
 77
 78        document = bs4.BeautifulSoup(response.text, 'html.parser')
 79
 80        element = document.select_one('input[name=setmode]')
 81        if (element is None):
 82            raise lms.backend.moodle.errors.MoodleAPIBreakageError()
 83
 84        context_str = element.get('data-context', None)
 85        if (context_str is None):
 86            raise lms.backend.moodle.errors.MoodleAPIBreakageError()
 87
 88        if (not isinstance(context_str, str)):
 89            raise lms.backend.moodle.errors.MoodleAPIBreakageError()
 90
 91        context = int(context_str)
 92
 93        params = {
 94            'sesskey': sesskey,
 95            'info': 'core_change_editmode',
 96        }
 97
 98        data = [
 99            {
100                'index': 0,
101                'methodname': 'core_change_editmode',
102                'args': {
103                    'setmode': True,
104                    'context': context,
105                },
106            }
107        ]
108
109        response, _ = edq.net.request.make_post(
110            f"{self.server}/lib/ajax/service.php",
111            additional_requests_options = {'params': params},
112            data = json.dumps(data),
113            headers = self.get_standard_headers(),
114        )
115
116        response, _ = edq.net.request.make_get(url, headers = self.get_standard_headers(), **kwargs)
117
118        return response
119
120    def reset_connection(self) -> None:
121        self._session_headers = None
122
123    def get_standard_headers(self, write: bool = False) -> typing.Dict[str, str]:
124        headers = super().get_standard_headers(write)
125
126        if (self._session_headers is not None):
127            headers.update(self._session_headers)
128
129        return headers
130
131    def _parse_cookies(self, response: requests.Response) -> typing.Dict[str, typing.Any]:
132        """
133        Parse Moodle cookies.
134        Return fake cookies when testing.
135        """
136
137        if (self.is_testing()):
138            return {
139                'moodlesession': 'testing-moodle-session',
140                'moodleid1_': 'testing-moodle-id',
141            }
142
143        return lms.util.net.parse_cookies(response.headers.get('set-cookie', None))
144
145    def _login(self, update_server: bool = True) -> None:
146        """
147        Try to login to the Moodle server.
148        If `update_server` is true, then this may try to update the backend's server location if redirected by the Moodle server.
149        """
150
151        # Check if we are already logged in.
152        if (self._session_headers is not None):
153            return
154
155        response, body = edq.net.request.make_get(self.server + '/login/index.php')
156        cookies = self._parse_cookies(response)
157
158        new_cookies = {
159            'MoodleSession': cookies['moodlesession'],
160        }
161        text_cookies = '; '.join(['='.join(items) for items in new_cookies.items()])
162
163        # Parse the login token from the page HTML.
164        document = bs4.BeautifulSoup(body, 'html.parser')
165        token = document.select('input[name="logintoken"]')[0]['value']
166
167        headers = {
168            'cookie': text_cookies,
169            'host': urllib.parse.urlparse(self.server).netloc,
170        }
171
172        data = {
173            'logintoken': token,
174            'username': self.auth_user,
175            'password': self.auth_password,
176        }
177
178        response, _ = edq.net.request.make_post(self.server + '/login/index.php',
179                headers = headers, data = data,
180                allow_redirects = False)
181
182        # Check for a successful login.
183        cookies = self._parse_cookies(response)
184        if ('moodleid1_' in cookies):
185            self._session_headers = {
186                'cookie': response.headers.get('set-cookie', None),
187                # Insert a header to identify the user.
188                'edq-lms-moodle-user': self.auth_user,
189            }
190
191            return
192
193        # Login Failed
194
195        # The specified server/host needs to match exactly what the Moodle server wants it to be,
196        # e.g., `127.0.0.1` does not work when the server wants the host to be `localhost`.
197        # If these do not match, we will get a redirect here.
198        # Use this redirect to discover the correct server.
199        location = response.headers.get('location', None)
200        if (update_server and (location is not None) and (not location.startswith(self.server))):
201            parts = urllib.parse.urlparse(location)
202            host = f"{parts.scheme}://{parts.netloc}"
203
204            _logger.debug(("Mismatch in the client-specified server ('%s') and server-requested host ('%s')."
205                    + " To avoid extra requests, update the server (e.g., `--server`) to match the host."),
206                    self.server, host)
207
208            # Update the server and try to login again (without updating the server again (to avoid loops)).
209            self.server = host
210            self._login(update_server = False)
211            return
212
213        raise ValueError(f"Could not log into Moodle server ({self.server}) with user '{self.auth_user}'. Is username/password correct?")
214
215    def courses_list(self,
216            **kwargs: typing.Any) -> typing.List[lms.model.courses.Course]:
217        self._login()
218
219        url = self.server + "/user/profile.php"
220        response, _ = edq.net.request.make_get(url, headers = self.get_standard_headers())
221
222        document = bs4.BeautifulSoup(response.text, 'html.parser')
223        cards = document.select('div.card-body')
224
225        node = None
226        for card in cards:
227            text = card.get_text()
228            if (text.startswith("Course details")):
229                node = card
230                break
231
232        if (node is None):
233            return []
234
235        links = node.select('a')
236
237        courses = []
238        for link in links:
239            name = link.get_text()
240
241            href = link.get('href', None)
242            if (href is None):
243                continue
244
245            id = str(href).rsplit("=", maxsplit = 1)[-1]
246
247            courses.append(lms.model.courses.Course(
248                id = id,
249                name = name,
250            ))
251
252        return sorted(courses)
253
254    def courses_users_list(self,
255            course_id: str,
256            **kwargs: typing.Any) -> typing.List[lms.model.users.CourseUser]:
257        self._login()
258
259        url = f"{self.server}/user/index.php?id={course_id}&perpage={RESULTS_PER_PAGE}"
260        response, _ = edq.net.request.make_get(url, headers = self.get_standard_headers())
261
262        document = bs4.BeautifulSoup(response.text, 'html.parser')
263
264        headers = document.select('table#participants thead tr th')
265        # { course_user_attribute (e.g. name): column class, ... }
266        classes = {}
267        for header in headers:
268            column_classes = header.get('class', None)
269            if (column_classes is None):
270                continue
271
272            # Parse and store the column's class (e.g. "c0").
273            # This class is referenced when storing corresponding course user data.
274            if (isinstance(column_classes, str)):
275                column_class = column_classes
276            else:
277                if ('header' in column_classes):
278                    column_classes.remove('header')
279
280                if (len(column_classes) != 1):
281                    continue
282
283                column_class = column_classes[0]
284
285            elements = header.select('div.commands a')
286            for element in elements:
287                attribute = element.get('data-column', None)
288                if (attribute is None):
289                    continue
290
291                classes[attribute] = column_class
292
293        rows = document.select('table#participants tbody tr:not(.emptyrow)')
294
295        users = []
296        for row in rows:
297            try:
298                id = row.select_one('.cell input[type="checkbox"]').get('id', None).removeprefix('user')  # type: ignore[union-attr]
299                name = row.select_one(f'.cell.{classes["fullname"]} a span').get('title', None).removeprefix('__EMPTY_NAME__ ')  # type: ignore[union-attr] # pylint: disable=line-too-long
300                email = row.select_one(f'.cell.{classes["email"]}').get_text()  # type: ignore[union-attr]
301                raw_role = row.select_one(f'.cell.{classes["roles"]} span a').get_text().strip().lower()  # type: ignore[union-attr]
302            except AttributeError as ex:
303                raise lms.backend.moodle.errors.MoodleAPIBreakageError() from ex
304
305            # HACK(JK): Moodle does not allow the Guest role when loading test data, so we patch the guest role during testing.
306            if (email == 'course-other@test.edulinq.org'):
307                raw_role = "guest"
308
309            users.append(lms.model.users.CourseUser(
310                id = id,
311                name = name,
312                email = email,
313                raw_role = raw_role,
314                role = ROLE_MAPPING.get(raw_role, None),
315            ))
316
317        return users
318
319    def courses_assignments_list(self,
320            course_id: str,
321            **kwargs: typing.Any) -> typing.List[lms.model.assignments.Assignment]:
322        self._login()
323
324        url = f"{self.server}/grade/report/grader/index.php?id={course_id}"
325
326        # Attempt to enable edit mode on the gradebook page.
327        response = self._get_edit_mode_page(url, enable_edit_mode = True)
328
329        # We expect graders to have edit mode access.
330        # If the edit mode request fails, we fall back to the non-grader gradebook page.
331        if (response is not None):
332            return self._fetch_assignments_grader(response)
333
334        return self._fetch_assignments_non_grader(course_id)
335
336    def _fetch_assignments_grader(self, response: requests.Response) -> typing.List[lms.model.assignments.Assignment]:
337        """
338        Fetch assignment data for users with grader permissions.
339        """
340
341        assignments = []
342
343        document = bs4.BeautifulSoup(response.text, 'html.parser')
344
345        activities = document.select('table#user-grades th.item')
346        for activity in activities:
347            # Parse and store the column's class (e.g. "c0").
348            target_class = None
349
350            column_classes = activity.get('class', None)
351            if (column_classes is None):
352                raise lms.backend.moodle.errors.MoodleAPIBreakageError()
353
354            for column_class in column_classes:
355                if (re.search(r'^c\d+$', column_class) is not None):
356                    target_class = column_class
357                    break
358
359            try:
360                id = str(activity.get('data-itemid', None))
361                name = str(activity.select_one('a.gradeitemheader').get_text())  # type: ignore[union-attr]
362
363                points_possible_str = document.select_one(f'td.{target_class} input').get('max', None)  # type: ignore[union-attr]
364                if (not isinstance(points_possible_str, str)):
365                    points_possible = 0.0
366                else:
367                    points_possible = float(points_possible_str)
368            except AttributeError as ex:
369                raise lms.backend.moodle.errors.MoodleAPIBreakageError() from ex
370
371            assignments.append(lms.model.assignments.Assignment(
372                id = id,
373                name = name,
374                points_possible = points_possible,
375            ))
376
377        return assignments
378
379    def _fetch_assignments_non_grader(self, course_id: str) -> typing.List[lms.model.assignments.Assignment]:
380        """
381        Fetch assignment data for users without grader permissions.
382        """
383
384        assignments = []
385
386        url = f"{self.server}/grade/report/user/index.php?id={course_id}"
387        response, _ = edq.net.request.make_get(url, headers = self.get_standard_headers())
388
389        document = bs4.BeautifulSoup(response.text, 'html.parser')
390
391        activities: typing.List[bs4.Tag] = list(document.find_all('tr[class]:not([class=""]):not(.spacer):not(.lastrow)'))
392        for activity in activities:
393            try:
394                id = str(activity.select_one('th').get('id', None).split('_')[1])  # type: ignore[union-attr]
395                name = str(activity.select_one('th a').get_text())  # type: ignore[union-attr]
396                points_possible = float(activity.select_one('td.column-range').get_text().split('–')[1])  # type: ignore[union-attr]
397            except AttributeError as ex:
398                raise lms.backend.moodle.errors.MoodleAPIBreakageError() from ex
399
400            assignments.append(lms.model.assignments.Assignment(
401                id = id,
402                name = name,
403                points_possible = points_possible,
404            ))
405
406        return assignments