"""Real routes, disposable files/SQLite, real JWTs; no network/AI calls.""" from datetime import datetime from pathlib import Path from tempfile import TemporaryDirectory import unittest from unittest.mock import AsyncMock, Mock, patch from sqlalchemy.dialects import postgresql import test_quiz_builder as builder from app.config import settings from app.models.attempt import QuizAttempt from app.models.email_verification import EmailVerification from app.models.flashcard import Flashcard, FlashcardDeck from app.models.pdf_document import PDFDocument from app.models.question import Question from app.models.quiz import Quiz from app.routers import teach, uploads, flashcards from app.utils.auth import create_access_token, get_current_user class PrivacyTests(unittest.TestCase): def setUp(self): builder.BuilderTests.setUp(self) self.tmp = TemporaryDirectory() self.settings_patch = patch.object(settings, 'UPLOAD_DIR', self.tmp.name) self.settings_patch.start() self.client.app.include_router(teach.router, prefix='/teach') self.client.app.include_router(uploads.router) self.client.app.include_router(flashcards.router, prefix='/flashcards') del self.client.app.dependency_overrides[get_current_user] for q in self.db.query(Question): q.image_path = f'questions/stem-{q.id}.png' q.explanation_image_path = f'questions/answer-{q.id}.png' self.file(q.image_path) self.file(q.explanation_image_path) self.db.commit() self.login(self.owner) self.quota = patch.object(teach, 'check_rate_limit').start() self.model = patch.object(teach, '_get_teach_model', return_value=('synthetic', None)).start() # Every question here has a figure, and the tutor now hands those to # the model. This file promises no network, so the capability lookup # is answered here rather than by asking the proxy. self.can_see = patch.object(teach.vision_service, 'can_see', return_value=True).start() self.find_similar = teach._find_similar_questions self.similar = patch.object(teach, '_find_similar_questions', return_value=[]).start() self.ai = AsyncMock(return_value='Tutor reply\n> Follow up') patch('app.services.ai_service.achat', new=self.ai).start() self.embedding = patch('app.services.embedding_service.embed_question').start() def tearDown(self): patch.stopall() self.tmp.cleanup() builder.BuilderTests.tearDown(self) def login(self, user, cookie=False): self.client.headers.pop('authorization', None) self.client.cookies.clear() token = create_access_token({'sub': user.email}) if cookie: self.client.cookies.set('pedshub_media', token, path='/uploads') else: self.client.headers['Authorization'] = f'Bearer {token}' def file(self, path, data=b'synthetic-image'): target = Path(self.tmp.name) / path target.parent.mkdir(parents=True, exist_ok=True) target.write_bytes(data) return target def chat(self, qid, aid=None): return self.client.post('/teach/chat', json={'question_id': qid, 'attempt_id': aid, 'messages': [{'role': 'user', 'content': 'Explain'}]}) def attempt(self, mode='study', completed=False, ids=None, user=1, quiz=2): a = QuizAttempt(user_id=user, quiz_id=quiz, mode=mode, selected_question_ids=ids, completed_at=datetime.utcnow() if completed else None) self.db.add(a) self.db.commit() return a def test_tutor_denial_precedes_all_external_boundaries(self): self.assertIn(self.chat(999).status_code, (403, 404)) for boundary in (self.quota, self.model, self.similar, self.ai): boundary.assert_not_called() # The tutor is handed the correct answer and told it may explain it, so # outside a session it is an answer key. Being able to reach a question # in the bank is not the same as having earned its answer: without an # attempt it opens for the people who write the question, and nobody # else — and it refuses before the quota, the model or the AI is asked. for qid in (1, 4, 5): self.assertEqual(self.chat(qid).status_code, 403) for boundary in (self.quota, self.model, self.similar, self.ai): boundary.assert_not_called() # Question 3 used to open here because this learner wrote it. Nothing # in the bank belongs to a person now, so authorship buys nothing and # it is refused like the rest. self.assertEqual(self.chat(3).status_code, 403) self.login(self.mod) self.assertEqual(self.chat(4).status_code, 200) # The context search is run as the caller, not as whoever wrote it. self.assertEqual(self.similar.call_args.args[2].id, self.mod.id) self.assertEqual(self.similar.call_args.args[1].id, 4) def test_tutor_attempt_modes_pool_and_review(self): # The tutor's site switch lives in Redis, which these tests share with # the running site — so an administrator turning the tutor off in the # interface used to turn this test red. What is under test here is who # may reach the tutor when it is *on*, so it says so. from app.services import site_settings site_settings.set_flag("tutor_in_quiz", True) self.addCleanup(site_settings.set_flag, "tutor_in_quiz", False) for mode in ('exam', None): a = self.attempt(mode=mode, ids=[5]) self.assertEqual(self.chat(5, a.id).status_code, 403) for a in (self.attempt(ids=[]), self.attempt(ids=[1]), self.attempt(user=2, ids=[5]), self.attempt(quiz=1, ids=[5])): self.assertEqual(self.chat(5, a.id).status_code, 403) self.assertEqual(self.chat(5, 999).status_code, 403) self.ai.assert_not_called() for a in (self.attempt(ids=[5]), self.attempt(mode='exam', completed=True, ids=[5])): self.assertEqual(self.chat(5, a.id).status_code, 200) def test_similarity_sql_filters_before_limit_and_full_prompt(self): q = self.db.get(Question, 1) q.options = [f'Option {i}' for i in range(8)] q.correct_answer = 'Option 2' q.question_text = 'Long stem ' * 500 prompt = teach._build_system_prompt(q, []) self.assertIn(q.question_text, prompt) # Lettered, matching the player: a tutor saying "option 3" about the # line the student sees as C leaves them reconciling two labellings. self.assertIn('H) Option 7', prompt) self.assertIn(q.explanation, prompt) # The keyed answer is marked against its own option, not only quoted # underneath — given the text alone a model matches the string back to # the wrong line when two options start the same way. self.assertIn('C) Option 2 <-- CORRECT ANSWER', prompt) # And said as a rule, so a model that would have answered differently # does not tell the student the marked answer is wrong. self.assertIn('answer key', prompt) self.assertIn('authoritative', prompt) # Compile actual eligibility/ranking SQL; mock only terminal execution. q.embedding = [0.1] * Question.__table__.c.embedding.type.dim captured = [] with patch('sqlalchemy.orm.Query.all', autospec=True, side_effect=lambda query: captured.append(query) or []): self.assertEqual(self.find_similar(self.db, q, self.owner), []) sql = str(captured[-1].statement.compile(dialect=postgresql.dialect())) # The predicate no longer consults a per-question flag; what it still # excludes is a deleted question. self.assertIn('questions.deleted_at IS NULL', sql) self.assertIn('<=>', sql) self.assertLess(sql.index('WHERE'), sql.index('ORDER BY')) self.assertLess(sql.index('ORDER BY'), sql.index('LIMIT')) with patch('sqlalchemy.orm.Query.all', side_effect=RuntimeError('Unavailable embeddings')): self.assertEqual(self.find_similar(self.db, q, self.owner), []) q.embedding = None self.assertEqual(self.find_similar(self.db, q, self.owner), []) self.db.rollback() def test_native_cookie_bearer_revocation_and_api_exclusion(self): self.client.headers.clear() self.assertEqual(self.client.get('/uploads/questions/stem-4.png').status_code, 401) self.login(self.owner, cookie=True) # Stems are readable in the bank. Answer images are not: 'answer-3' # used to open because this learner wrote question 3, and authorship # is no longer a claim the bank honours. for path in ('stem-1', 'stem-3', 'stem-4'): res = self.client.get(f'/uploads/questions/{path}.png') self.assertEqual(res.status_code, 200, res.text) self.assertEqual(res.headers['cache-control'], 'private, no-store') self.assertEqual(res.headers['vary'], 'Cookie, Authorization') # Somebody else's answer, with no attempt behind it: refused. for answer in ('answer-1', 'answer-3'): self.assertEqual( self.client.get(f'/uploads/questions/{answer}.png').status_code, 404, answer) # A derivative may be kept by the browser that asked for it — never by # a shared cache, which in front of access-controlled images is how one # learner is served another's private figure. The original still says # no-store; only the thumbnail, which cannot change, may be kept. # A real image, because a derivative only exists where one can be made: # the other fixtures here are the bytes `synthetic-image`, and asking # for a thumbnail of those correctly gets the original back. import io from PIL import Image as PILImage buffer = io.BytesIO() PILImage.new('RGB', (900, 600), (200, 30, 30)).save(buffer, format='PNG') self.file('questions/stem-1.png', buffer.getvalue()) thumb = self.client.get('/uploads/questions/stem-1.png?w=256') self.assertEqual(thumb.status_code, 200, thumb.text) self.assertIn('private', thumb.headers['cache-control']) self.assertIn('max-age=', thumb.headers['cache-control']) self.assertNotIn('public', thumb.headers['cache-control']) self.assertEqual(thumb.headers['vary'], 'Cookie, Authorization') self.db.get(Question, 1).deleted_at = datetime(2026, 1, 1) self.db.commit() self.assertEqual(self.client.get('/uploads/questions/stem-1.png').status_code, 404) self.login(self.owner) self.assertEqual(self.client.get('/uploads/questions/stem-3.png').status_code, 200) self.db.add(EmailVerification(user_id=1, token='synthetic', expires_at=datetime(2099, 1, 1))) self.db.commit() self.assertEqual(self.client.get('/uploads/questions/stem-3.png').status_code, 403) self.assertEqual(self.chat(3).status_code, 403) def test_an_attempt_reaches_only_the_questions_it_actually_holds(self): # An attempt id in the address is not a skeleton key: it opens the # questions that attempt was given, to the learner it belongs to, and # nothing else. (The stem itself is bank content and needs no attempt — # the answer beside it is reachable the same way, which is the tutor's # rule too, so a session is not what protects it.) a = self.attempt(mode='study', ids=[5]) url = f'/uploads/questions/answer-5.png?attempt_id={a.id}' self.assertEqual(self.client.get(url).status_code, 200) a.selected_question_ids = [] self.db.commit() self.assertEqual(self.client.get(url).status_code, 404) a.selected_question_ids = [5] self.db.commit() self.login(self.peer) self.assertEqual(self.client.get(url).status_code, 404) self.login(self.owner) self.assertEqual(self.client.get(f'/uploads/questions/answer-5.png?attempt_id=99999').status_code, 404) def test_extraction_original_pdf_unattached_and_cards(self): self.db.add_all([PDFDocument(id=1, user_id=2, filename='private.pdf', original_filename='source.pdf'), PDFDocument(id=2, user_id=1, filename='own.pdf', original_filename='own.pdf')]) for path in ('private.pdf', 'own.pdf', 'images/doc_1/shared.png', 'images/doc_1/adjacent.png', 'images/doc_2/own.png', 'questions/1/draft.png', 'questions/2/draft.png', 'questions/legacy.png'): self.file(path) self.db.get(Question, 1).image_path = 'images/doc_1/shared.png' self.db.get(Question, 2).image_path = 'private.pdf' self.db.commit() for path, code in [('private.pdf', 404), ('own.pdf', 200), ('images/doc_1/shared.png', 200), ('images/doc_1/adjacent.png', 404), ('images/doc_2/own.png', 200), ('questions/1/draft.png', 200), ('questions/2/draft.png', 404), ('questions/legacy.png', 404)]: self.assertEqual(self.client.get('/uploads/' + path).status_code, code, path) deck = FlashcardDeck(user_id=2, title='Cards', is_shared=0) self.db.add(deck) self.db.flush() self.db.add(Flashcard(deck_id=deck.id, front='front', back='back', image_path='questions/2/draft.png')) self.db.commit() self.assertEqual(self.client.get('/uploads/questions/2/draft.png').status_code, 404) deck.is_shared = 1 self.db.commit() self.assertEqual(self.client.get('/uploads/questions/2/draft.png').status_code, 200) deck.deleted_at = datetime.utcnow() self.db.commit() self.assertEqual(self.client.get('/uploads/questions/2/draft.png').status_code, 404) self.login(self.mod) self.assertEqual(self.client.get('/uploads/questions/legacy.png').status_code, 200) def test_attachment_forgery_atomic_and_chooser(self): payload = {'question_text': 'Created', 'options': ['yes', 'no'], 'correct_answer': 'yes'} before = self.db.query(Question).count() # Learners cannot manage questions at all. for field in ('image_path', 'explanation_image_path'): self.assertEqual(self.client.post('/questions/create', json={**payload, field: 'questions/stem-1.png'}).status_code, 403) self.assertEqual(self.client.patch('/questions/3', json={'question_text': 'Forged', field: 'questions/stem-1.png'}).status_code, 403) self.assertEqual(self.db.query(Question).count(), before) self.assertEqual(self.db.get(Question, 3).question_text, 'Question 3') self.login(self.mod) # Educator validation: unsafe and malformed paths are rejected atomically. for field in ('image_path', 'explanation_image_path'): for path in ('../escape', '/uploads/questions/../escape', '//other/asset'): res = self.client.post('/questions/create', json={**payload, field: path}) self.assertIn(res.status_code, (400, 403), res.text) self.assertEqual(self.db.query(Question).count(), before) res = self.client.patch('/questions/3', json={'question_text': 'Forged', field: path}) self.assertIn(res.status_code, (400, 403), res.text) self.assertEqual(self.db.get(Question, 3).question_text, 'Question 3') # Moderators may reference peer questions; that is a moderator privilege, not a forgery. res = self.client.post('/questions/create', json={**payload, 'image_path': '/uploads/questions/stem-4.png'}) self.assertEqual(res.status_code, 200, res.text) res = self.client.post('/questions/create', json={**payload, 'image_path': '/uploads/questions/stem-1.png'}) self.assertEqual(res.status_code, 200, res.text) self.assertEqual(res.json()['image_path'], 'questions/stem-1.png') paths = [p['image_path'] for p in self.client.get('/questions/images').json()] self.assertEqual(paths, sorted(set(paths))) self.assertIn('questions/stem-1.png', paths) self.assertIn('questions/stem-4.png', paths) # Now bank-visible via the created shared question. self.login(self.mod) for field in ('image_path', 'explanation_image_path'): for value in ([], {}, 17, '../escape'): res = self.client.patch('/quizzes/1/questions/1', json={'question_text': 'Forged', field: value}) self.assertEqual(res.status_code, 400, res.text) self.assertEqual(self.db.get(Question, 1).question_text, 'Question 1') self.embedding.assert_called() # One embedding per created question (two created here). def test_same_site_urls_share_path_authorization_and_legacy_references(self): with patch.object(settings, 'APP_URL', 'https://app.example.com'): payload = {'question_text': 'Absolute image', 'options': ['yes', 'no'], 'correct_answer': 'yes'} count = self.db.query(Question).count() for url in ('https://app.example.com/uploads/questions/stem-4.png', 'http://app.example.com/uploads/questions/answer-4.png'): response = self.client.post('/questions/create', json={**payload, 'image_path': url}) self.assertEqual(response.status_code, 403, response.text) self.assertEqual(self.db.query(Question).count(), count) response = self.client.patch('/questions/3', json={'question_text': 'Forged', 'explanation_image_path': url}) self.assertEqual(response.status_code, 403, response.text) self.assertEqual(self.db.get(Question, 3).question_text, 'Question 3') self.login(self.mod) response = self.client.post('/questions/create', json={**payload, 'image_path': 'https://app.example.com/uploads/questions/stem-1.png'}) self.assertEqual(response.status_code, 200, response.text) self.assertEqual(response.json()['image_path'], 'questions/stem-1.png') for url in ('https://external.example:bad/a.png', 'https://external.example:99999/a.png'): response = self.client.patch('/questions/3', json={'image_path': url}) self.assertEqual(response.status_code, 400, response.text) response = self.client.patch('/questions/3', json={'image_path': 'https://external.example/figure.png'}) self.assertEqual(response.status_code, 200, response.text) self.assertEqual(response.json()['image_path'], 'https://external.example/figure.png') self.db.get(Question, 2).image_path = 'https://app.example.com/uploads/legacy-public.svg' self.file('legacy-public.svg') self.db.commit() self.client.headers.clear() self.assertEqual(self.client.get('/uploads/legacy-public.svg').status_code, 401) self.login(self.owner) self.assertEqual(self.client.get('/uploads/legacy-public.svg').status_code, 200) self.db.get(Question, 2).deleted_at = datetime(2026, 1, 1) self.db.commit() self.assertEqual(self.client.get('/uploads/legacy-public.svg').status_code, 404) self.db.get(Question, 5).image_path = 'http://app.example.com/uploads/questions/stem-5.png' self.db.commit() self.assertEqual(self.client.get('/uploads/questions/stem-5.png').status_code, 200) def test_card_only_moderator_denial_admin_success_and_question_grant(self): admin = builder.User(id=4, name='Admin', email='admin@example.com', hashed_password='unused', role='admin') deck = FlashcardDeck(user_id=2, title='Private peer cards', is_shared=0) self.db.add_all([admin, deck]) self.db.flush() card = Flashcard(deck_id=deck.id, front='Private', back='Private', image_path='cards/private.png') self.db.add(card) self.db.commit() payload = {'question_text': 'Copy', 'options': ['yes', 'no'], 'correct_answer': 'yes'} for path in ('cards/private.png', 'questions/2/card-only.png'): card.image_path = path self.file(path) self.db.commit() self.login(self.mod) # A deck is the bank's, so an educator reaches it. The image on # the card is a separate question, and that is the boundary this # test is about: it stays refused. self.assertEqual(self.client.get(f'/flashcards/{deck.id}').status_code, 200) self.assertEqual(self.client.get('/uploads/' + path).status_code, 404) before = self.db.query(Question).count() response = self.client.post('/questions/create', json={**payload, 'image_path': path}) self.assertEqual(response.status_code, 403, response.text) self.assertEqual(self.db.query(Question).count(), before) response = self.client.patch('/questions/1', json={'explanation_image_path': path, 'question_text': 'Forged'}) self.assertEqual(response.status_code, 403, response.text) self.assertEqual(self.db.get(Question, 1).question_text, 'Question 1') self.login(admin) self.assertEqual(self.client.get(f'/flashcards/{deck.id}').status_code, 200) self.assertEqual(self.client.get('/uploads/' + path).status_code, 200) self.assertEqual(self.client.post('/questions/create', json={**payload, 'image_path': path}).status_code, 200) # Even inside extraction directories, a card reference stays on card ACL. self.login(self.mod) card.image_path = 'images/doc_1/card-only.png' self.file(card.image_path) self.db.commit() self.assertEqual(self.client.get('/uploads/images/doc_1/card-only.png').status_code, 404) self.assertEqual(self.client.patch('/questions/1', json={'image_path': card.image_path}).status_code, 403) self.login(admin) self.assertEqual(self.client.get('/uploads/images/doc_1/card-only.png').status_code, 200) # A real question reference independently gives its moderator access. self.login(self.mod) self.db.get(Question, 4).image_path = 'cards/moderated-question.png' card.image_path = 'cards/moderated-question.png' self.file(card.image_path) self.db.commit() self.assertEqual(self.client.get('/uploads/cards/moderated-question.png').status_code, 200) self.assertEqual(self.client.post('/questions/create', json={**payload, 'image_path': card.image_path}).status_code, 200) def test_legacy_aliases_are_canonical_for_classification_and_permissions(self): with patch.object(settings, 'APP_URL', 'https://app.example.com'): self.file('legacy-private.svg') self.file('course_files/card-only.svg') question = self.db.get(Question, 3) variants = [ 'https://APP.EXAMPLE.COM/uploads/legacy-private.svg?v=1#view', 'http://app.example.com:80/uploads/legacy-private.svg#part', '/uploads/legacy-private.svg?size=1', 'legacy-private.svg#part', 'https://app%2Eexample.com/uploads/legacy-private.svg', '/uploads/images/../legacy-private.svg', 'https://app.example.com/uploads/%6Cegacy-private.svg', r'https://app.example.com\uploads\legacy-private.svg?v=1', ] for value in variants: question.image_path = value self.db.commit() self.client.headers.clear() self.client.cookies.clear() self.assertEqual(self.client.get('/uploads/legacy-private.svg?v=1').status_code, 401, value) self.login(self.owner) self.assertEqual(self.client.get('/uploads/legacy-private.svg?v=1').status_code, 200, value) self.login(self.peer) # The question is in the bank, so its image is reachable by # anybody who may sit it — which, with per-question sharing # gone, is everybody. self.assertEqual(self.client.get('/uploads/legacy-private.svg').status_code, 200, value) self.login(self.owner) bank_paths = self.client.get('/questions/images').json() self.assertIn({'image_path': 'legacy-private.svg', 'url': '/uploads/legacy-private.svg'}, bank_paths) # Parsing a browser-normalized URL must not bypass checks on new writes. self.login(self.peer) response = self.client.patch('/questions/4', json={'image_path': 'https://app%2Eexample.com/uploads/legacy-private.svg'}) self.assertEqual(response.status_code, 403) # Learners cannot edit questions. self.login(self.mod) for value in ('https://app%2Eexample.com/uploads/legacy-private.svg', r'https://app.example.com\uploads\legacy-private.svg'): response = self.client.patch('/questions/4', json={'image_path': value}) self.assertEqual(response.status_code, 400, response.text) # Even a legacy-LMS directory becomes protected when a private card references it. deck = FlashcardDeck(user_id=1, title='Private owner cards', is_shared=0) self.db.add(deck) self.db.flush() self.db.add(Flashcard(deck_id=deck.id, front='Private', back='Private', image_path='https://APP.EXAMPLE.COM/uploads/course_files/card-only.svg?v=1#card')) self.db.commit() self.client.headers.clear() self.assertEqual(self.client.get('/uploads/course_files/card-only.svg').status_code, 401) self.login(self.owner) self.assertEqual(self.client.get('/uploads/course_files/card-only.svg').status_code, 200) self.login(self.mod) self.assertEqual(self.client.get('/uploads/course_files/card-only.svg').status_code, 404) self.assertEqual(self.client.patch('/questions/1', json={'image_path': 'course_files/card-only.svg'}).status_code, 403) def test_orphaned_files_do_not_become_anonymous_after_reference_deletion(self): self.file('orphan.svg') question = self.db.get(Question, 3) question.image_path = 'orphan.svg' self.db.commit() self.assertEqual(self.client.get('/uploads/orphan.svg').status_code, 200) self.db.delete(question) self.db.commit() self.client.headers.clear() self.assertEqual(self.client.get('/uploads/orphan.svg').status_code, 401) self.login(self.peer) self.assertEqual(self.client.get('/uploads/orphan.svg').status_code, 404) for prefix in ('course_files', 'course_thumbnails', 'scorm', 'certificates'): self.file(f'{prefix}/unrelated.bin') self.client.headers.clear() self.assertEqual(self.client.get(f'/uploads/{prefix}/unrelated.bin').status_code, 200) def test_paths_svg_head_range_upload_and_legacy_policy(self): res = self.client.post('/questions/upload-image', files={'file': ('test.svg', b'', 'image/svg+xml')}) self.assertEqual(res.status_code, 200, res.text) self.assertTrue(res.json()['image_path'].startswith('questions/1/')) url = res.json()['url'] res = self.client.get(url) self.assertEqual(res.headers['x-content-type-options'], 'nosniff') self.assertIn('sandbox', res.headers['content-security-policy']) head = self.client.head(url) self.assertEqual(head.status_code, 200) self.assertEqual(head.content, b'') self.assertEqual(head.headers['content-length'], str(len(res.content))) ranged = self.client.get(url, headers={'Range': 'bytes=0-3'}) # HTTP permits ignoring Range; pinned Starlette returns a full 200 here. # Production byte-range delivery is checked through Nginx's native filter. self.assertIn(ranged.status_code, (200, 206)) self.assertEqual(ranged.content, res.content[:4] if ranged.status_code == 206 else res.content) Path(self.tmp.name, 'questions', 'alias.png').symlink_to(Path(self.tmp.name, 'questions', 'stem-4.png')) Path(self.tmp.name, 'questions', 'outside.png').symlink_to('/etc/passwd') for path in ('questions/alias.png', 'questions/outside.png', 'questions/%2e%2e/escape', 'questions/%252e%252e/escape', 'questions/%5cescape', 'questions/missing.png', 'questions/' + 'x' * 300): self.assertEqual(self.client.get('/uploads/' + path).status_code, 404, path) self.file('scorm/test/index.html', b'Legacy SCORM') self.client.headers.clear() legacy = self.client.get('/uploads/scorm/test/index.html') self.assertEqual(legacy.status_code, 200) self.assertEqual(legacy.headers['cache-control'], 'private, no-store') if __name__ == '__main__': unittest.main()