fix: import unpublished exam rooms and add opt-in school HTML feedback

server/feedback/feedback_service.py

@@ -18,6 +18,7 @@ from dataclasses import dataclass, field
 from email.message import EmailMessage
 from email.policy import SMTP
 from email.utils import formatdate
+from html.parser import HTMLParser
 from pathlib import Path
 
 from fastapi import FastAPI, Request
@@ -30,7 +31,8 @@ MIB = 1024 * 1024
 FILE_LIMIT = 8 * MIB
 TOTAL_FILE_LIMIT = 20 * MIB
 LOG_LIMIT = 256 * 1024
-BODY_LIMIT = TOTAL_FILE_LIMIT + LOG_LIMIT + 64 * 1024
+HTML_LIMIT = 2 * MIB
+BODY_LIMIT = TOTAL_FILE_LIMIT + LOG_LIMIT + HTML_LIMIT + 64 * 1024
 ID_PATTERN = re.compile(r"[A-Za-z0-9_-]{16,128}\Z")
 ADDRESS_PATTERN = re.compile(r"[A-Za-z0-9._%+-]+@(?:[A-Za-z0-9-]+\.)+[A-Za-z]{2,63}\Z")
 
@@ -105,6 +107,95 @@ class ApiError(Exception):
         self.submission_id, self.status = submission_id, status
 
 
+class SchoolHtmlValidator(HTMLParser):
+    """Accept only the inert diagnostic schema emitted by the app, not web pages."""
+    tags = set(('html head meta title body input table thead tbody tfoot tr td th caption colgroup col '
+                'span div p b i strong em u small br hr a font center nobr label s sub sup pre code').split())
+    void = {'meta', 'input', 'br', 'hr', 'col'}
+    csp = "default-src 'none'; base-uri 'none'; form-action 'none'"
+
+    def __init__(self):
+        super().__init__(convert_charrefs=True)
+        self.stack, self.policy, self.charset, self.root_count = [], False, False, 0
+
+    def handle_starttag(self, tag, attrs):
+        values = dict(attrs)
+        if len(values) != len(attrs) or tag not in self.tags | {'script'}:
+            raise ValueError('tag')
+        if tag == 'html':
+            self.root_count += 1
+            if self.stack or self.root_count != 1:
+                raise ValueError('root')
+        elif not self.stack:
+            raise ValueError('outside root')
+        if tag == 'meta':
+            if self.stack[-1:] != ['head']:
+                raise ValueError('metadata location')
+            if values == {'charset': 'utf-8'}:
+                self.charset = True
+            elif values == {'http-equiv': 'Content-Security-Policy', 'content': self.csp}:
+                self.policy = True
+            else:
+                raise ValueError('metadata')
+        elif tag == 'input':
+            if (set(values) != {'type', 'name', 'value'} or values['type'] != 'hidden'
+                    or values['name'] not in {'semester.id', 'examType.id'}
+                    or not re.fullmatch(r'\d{1,6}', values['value'] or '')):
+                raise ValueError('input')
+        elif tag == 'script':
+            if values != {'type': 'application/x-tpschedule-course-data'}:
+                raise ValueError('script')
+        else:
+            for key, value in attrs:
+                if key == 'id' and tag == 'table' and value == 'manualArrangeCourseTable':
+                    continue
+                if key in {'rowspan', 'colspan'} and re.fullmatch(r'\d{1,2}', value or ''):
+                    continue
+                raise ValueError('attribute')
+        if tag not in self.void:
+            self.stack.append(tag)
+
+    def handle_endtag(self, tag):
+        if not self.stack or self.stack.pop() != tag:
+            raise ValueError('unbalanced')
+
+    def handle_startendtag(self, tag, attrs):
+        if tag not in self.void:
+            raise ValueError('self closing')
+        self.handle_starttag(tag, attrs)
+
+    def handle_data(self, data):
+        if not self.stack and data.strip():
+            raise ValueError('outside root')
+        if self.stack[-1:] == ['script'] and '<' in data:
+            raise ValueError('script delimiter')
+
+    def handle_comment(self, data):
+        if '[if' in data.lower() or '--' in data:
+            raise ValueError('conditional comment')
+
+    def handle_decl(self, decl):
+        if decl.lower() != 'doctype html':
+            raise ValueError('doctype')
+
+    def unknown_decl(self, data):
+        raise ValueError('declaration')
+
+    def handle_pi(self, data):
+        raise ValueError('processing instruction')
+
+    @classmethod
+    def validate(cls, text):
+        try:
+            parser = cls()
+            parser.feed(text)
+            parser.close()
+            if parser.stack or not parser.policy or not parser.charset or parser.root_count != 1:
+                raise ValueError('incomplete')
+        except (ValueError, AssertionError):
+            raise ApiError(415, 'unsupported_media', 'School HTML must use the inert diagnostic schema.') from None
+
+
 class SendFailure(Exception):
     """SMTP definitely did not accept this message. Retrying is safe."""
 
@@ -363,10 +454,15 @@ def safe_filename(filename, extension):
 
 async def parse_submission(request):
     try:
-        async with request.form(max_files=4, max_fields=5, max_part_size=20000) as form:
-            fields, uploads, diagnostic_upload = {}, [], None
+        async with request.form(max_files=5, max_fields=5, max_part_size=20000) as form:
+            fields, uploads, diagnostic_upload, html_upload = {}, [], None, None
             for name, value in form.multi_items():
                 if isinstance(value, UploadFile):
+                    if name == "diagnostic_html":
+                        if html_upload is not None:
+                            raise ApiError(422, "validation_error", "Only one school HTML is allowed.")
+                        html_upload = value
+                        continue
                     if name == "diagnostics":
                         if diagnostic_upload is not None:
                             raise ApiError(422, "validation_error", "Only one diagnostic log is allowed.")
@@ -393,7 +489,7 @@ async def parse_submission(request):
             if kind == "problem":
                 if not fields.get("message", "").strip() or "school" in fields:
                     raise ApiError(422, "validation_error", "Problem feedback requires message and must omit school.")
-            elif ("message" in fields or uploads or diagnostic_upload or not fields.get("school", "").strip()
+            elif ("message" in fields or uploads or diagnostic_upload or html_upload or not fields.get("school", "").strip()
                   or not fields.get("contact", "").strip()):
                 raise ApiError(422, "validation_error", "Development requests require school and contact; omit message and attachments.")
             attachments, total = [], 0
@@ -420,6 +516,19 @@ async def parse_submission(request):
                 if (diagnostic_upload.content_type or '').split(';')[0].strip().lower() != 'text/plain' or '\x00' in text or not text.startswith('TPSchedule diagnostic log\n'):
                     raise ApiError(415, "unsupported_media", "Invalid diagnostic log.")
                 attachments.append(('TPSchedule-diagnostics.txt', 'text/plain', data))
+            if html_upload is not None:
+                data = await html_upload.read(HTML_LIMIT + 1)
+                if len(data) > HTML_LIMIT:
+                    raise ApiError(413, "upload_too_large", "School HTML exceeds 2 MiB.")
+                try:
+                    text = data.decode('utf-8')
+                except UnicodeDecodeError:
+                    raise ApiError(415, "unsupported_media", "School HTML must be UTF-8.") from None
+                if ((html_upload.content_type or '').split(';')[0].strip().lower() != 'text/html'
+                        or '\x00' in text or not text.startswith('<!doctype html><!-- TPSchedule school diagnostic -->')):
+                    raise ApiError(415, "unsupported_media", "Invalid school HTML diagnostic.")
+                SchoolHtmlValidator.validate(text)
+                attachments.append(('TPSchedule-school-page.html', 'text/html', data))
             return fields, attachments
     except HTTPException as exc:
         detail = str(exc.detail).lower()