lms.backend.moodle.backend

  1# pylint: disable=abstract-method
  2
  3import logging
  4import typing
  5import urllib.parse
  6
  7import bs4
  8import edq.net.request
  9import requests
 10
 11import lms.model.backend
 12import lms.model.constants
 13import lms.util.net
 14
 15_logger = logging.getLogger(__name__)
 16
 17ROLE_MAPPING: typing.Dict[str, lms.model.users.CourseRole] = {
 18    "guest": lms.model.users.CourseRole.OTHER,
 19    "student": lms.model.users.CourseRole.STUDENT,
 20    "non-editing teacher": lms.model.users.CourseRole.GRADER,
 21    "teacher": lms.model.users.CourseRole.ADMIN,
 22    "manager": lms.model.users.CourseRole.OWNER,
 23}
 24
 25# Moodle shows 5000 users per page when asked to fetch all results.
 26RESULTS_PER_PAGE: int = 5000
 27
 28class MoodleBackend(lms.model.backend.APIBackend):
 29    """ An API backend for the Moodle LMS. """
 30
 31    def __init__(self,
 32            **kwargs: typing.Any) -> None:
 33        super().__init__(**kwargs)
 34
 35        assert(self.config.backend_type == lms.model.constants.BackendType.MOODLE)
 36
 37        if (self.config.auth_user is None):
 38            raise ValueError("Moodle backends require a username.")
 39
 40        self.auth_user: str = self.config.auth_user
 41        """
 42        The user to authenticate with.
 43        This is set in config and compied for type checking.
 44        """
 45
 46        if (self.config.auth_password is None):
 47            raise ValueError("Moodle backends require a password.")
 48
 49        self.auth_password: str = self.config.auth_password.cleartext
 50        """
 51        The (cleartext) password to authenticate with.
 52        This is set in config and compied for type checking.
 53        """
 54
 55        self._session_headers: typing.Union[typing.Dict[str, typing.Any], None] = None
 56        """ The headers (e.g., cookies) for our logged in Moodle session. """
 57
 58    def reset_connection(self) -> None:
 59        self._session_headers = None
 60
 61    def get_standard_headers(self, write: bool = False) -> typing.Dict[str, str]:
 62        headers = super().get_standard_headers(write)
 63
 64        if (self._session_headers is not None):
 65            headers.update(self._session_headers)
 66
 67        return headers
 68
 69    def _parse_cookies(self, response: requests.Response) -> typing.Dict[str, typing.Any]:
 70        """
 71        Parse Moodle cookies.
 72        Return fake cookies when testing.
 73        """
 74
 75        if (self.is_testing()):
 76            return {
 77                'moodlesession': 'testing-moodle-session',
 78                'moodleid1_': 'testing-moodle-id',
 79            }
 80
 81        return lms.util.net.parse_cookies(response.headers.get('set-cookie', None))
 82
 83    def _login(self, update_server: bool = True) -> None:
 84        """
 85        Try to login to the Moodle server.
 86        If `update_server` is true, then this may try to update the backend's server location if redirected by the Moodle server.
 87        """
 88
 89        # Check if we are already logged in.
 90        if (self._session_headers is not None):
 91            return
 92
 93        response, body = edq.net.request.make_get(self.server + '/login/index.php')
 94        cookies = self._parse_cookies(response)
 95
 96        new_cookies = {
 97            'MoodleSession': cookies['moodlesession'],
 98        }
 99        text_cookies = '; '.join(['='.join(items) for items in new_cookies.items()])
100
101        # Parse the login token from the page HTML.
102        document = bs4.BeautifulSoup(body, 'html.parser')
103        token = document.select('input[name="logintoken"]')[0]['value']
104
105        headers = {
106            'cookie': text_cookies,
107            'host': urllib.parse.urlparse(self.server).netloc,
108        }
109
110        data = {
111            'logintoken': token,
112            'username': self.auth_user,
113            'password': self.auth_password,
114        }
115
116        response, _ = edq.net.request.make_post(self.server + '/login/index.php',
117                headers = headers, data = data,
118                allow_redirects = False)
119
120        # Check for a successful login.
121        cookies = self._parse_cookies(response)
122        if ('moodleid1_' in cookies):
123            self._session_headers = {
124                'cookie': response.headers.get('set-cookie', None),
125                # Insert a header to identify the user.
126                'edq-lms-moodle-user': self.auth_user,
127            }
128            return
129
130        # Login Failed
131
132        # The specified server/host needs to match exactly what the Moodle server wants it to be,
133        # e.g., `127.0.0.1` does not work when the server wants the host to be `localhost`.
134        # If these do not match, we will get a redirect here.
135        # Use this redirect to discover the correct server.
136        location = response.headers.get('location', None)
137        if (update_server and (location is not None) and (not location.startswith(self.server))):
138            parts = urllib.parse.urlparse(location)
139            host = f"{parts.scheme}://{parts.netloc}"
140
141            _logger.debug(("Mismatch in the client-specified server ('%s') and server-requested host ('%s')."
142                    + " To avoid extra requests, update the server (e.g., `--server`) to match the host."),
143                    self.server, host)
144
145            # Update the server and try to login again (without updating the server again (to avoid loops)).
146            self.server = host
147            self._login(update_server = False)
148            return
149
150        raise ValueError(f"Could not log into Moodle server ({self.server}) with user '{self.auth_user}'. Is username/password correct?")
151
152    def courses_list(self,
153            **kwargs: typing.Any) -> typing.List[lms.model.courses.Course]:
154        self._login()
155
156        url = self.server + "/user/profile.php"
157        response, _ = edq.net.request.make_get(url, headers = self.get_standard_headers())
158
159        document = bs4.BeautifulSoup(response.text, 'html.parser')
160        cards = document.select('div.card-body')
161
162        node = None
163        for card in cards:
164            text = card.get_text()
165            if (text.startswith("Course details")):
166                node = card
167                break
168
169        if (node is None):
170            return []
171
172        links = node.select('a')
173
174        courses = []
175        for link in links:
176            name = link.get_text()
177
178            href = link.get('href', None)
179            if (href is None):
180                continue
181
182            id = str(href).rsplit("=", maxsplit = 1)[-1]
183
184            courses.append(lms.model.courses.Course(
185                id = id,
186                name = name,
187            ))
188
189        return sorted(courses)
190
191    def courses_users_list(self,
192            course_id: str,
193            **kwargs: typing.Any) -> typing.List[lms.model.users.CourseUser]:
194        self._login()
195
196        url = f"{self.server}/user/index.php?id={course_id}&perpage={RESULTS_PER_PAGE}"
197        response, _ = edq.net.request.make_get(url, headers = self.get_standard_headers())
198
199        document = bs4.BeautifulSoup(response.text, 'html.parser')
200
201        headers = document.select('table#participants thead tr th')
202        # { course_user_attribute (e.g. name): column class, ... }
203        classes = {}
204        for header in headers:
205            column_classes = header.get('class', None)
206            if (column_classes is None):
207                continue
208
209            # Parse and store the column's class (e.g. "c0").
210            # This class is referenced when storing corresponding course user data.
211            if (isinstance(column_classes, str)):
212                column_class = column_classes
213            else:
214                if ('header' in column_classes):
215                    column_classes.remove('header')
216
217                if (len(column_classes) != 1):
218                    continue
219
220                column_class = column_classes[0]
221
222            elements = header.select('div.commands a')
223            for element in elements:
224                attribute = element.get('data-column', None)
225                if (attribute is None):
226                    continue
227
228                classes[attribute] = column_class
229
230        rows = document.select('table#participants tbody tr:not(.emptyrow)')
231
232        users = []
233        for row in rows:
234            try:
235                id = row.select_one('.cell input[type="checkbox"]').get('id', None).removeprefix('user')  # type: ignore[union-attr]
236                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
237                email = row.select_one(f'.cell.{classes["email"]}').get_text()  # type: ignore[union-attr]
238                raw_role = row.select_one(f'.cell.{classes["roles"]} span a').get_text().strip().lower()  # type: ignore[union-attr]
239            except AttributeError as _:
240                _logger.warning("Unable to list users. Moodle data structure has changed. Contact project developers.")
241                continue
242
243            # HACK(JK): Moodle does not allow the Guest role when loading test data, so we patch the guest role during testing.
244            if (email == 'course-other@test.edulinq.org'):
245                raw_role = "guest"
246
247            users.append(lms.model.users.CourseUser(
248                id = id,
249                name = name,
250                email = email,
251                raw_role = raw_role,
252                role = ROLE_MAPPING.get(raw_role, None),
253            ))
254
255        return users