danvics/pdf-quiz-generator
1
0
Fork 0
Code Issues Pull requests Projects Releases 3 Packages Wiki Activity Actions
pdf-quiz-generator/backend/tests/test_related_privacy.py
Daniel 1ce3eec7cb feat: authorize tutor context and protect uploaded media 
Tutor questions require owned selected attempts; similarity context filters eligibility before ranking. Uploads move to a permission-aware boundary with reference ACLs, canonical legacy aliases, pre-mutation attachment checks and card-aware moderator rules. Nginx stops caching media and supplies native byte ranges. Verified 37 deployed-image backend tests, 69 frontend tests/build, real pgvector/Nginx/browser checks, and two independent reviews.
2026-09-07 15:04:00 +02:00

415 lines
24 KiB
Python
Raw Blame History

"""Real routes, disposable files/SQLite, real JWTs; no network/AI calls."""
from datetime import datetime
from pathlib import Path
from tempfile import TemporaryDirectory
from types import SimpleNamespace
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.course import CourseEnrollment
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()
self.find_similar = teach._find_similar_questions
self.similar = patch.object(teach, '_find_similar_questions', return_value=[]).start()
self.ai = patch('litellm.acompletion', new_callable=AsyncMock).start()
self.ai.return_value = SimpleNamespace(choices=[SimpleNamespace(message=SimpleNamespace(content='Tutor reply\n> Follow up'))])
self.embedding = patch('app.services.embedding_service.embed_question').start()
self.proxy = patch('app.services.ai_service._proxy_model', return_value='synthetic').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 enroll(self):
self.db.add(CourseEnrollment(course_id=1, user_id=1))
self.db.get(Quiz, 2).allow_review = 1
self.db.commit()
def test_tutor_denial_precedes_all_external_boundaries(self):
for qid in (4, 5, 999):
self.assertIn(self.chat(qid).status_code, (403, 404))
for boundary in (self.quota, self.model, self.similar, self.ai):
boundary.assert_not_called()
for qid in (1, 3):
res = self.chat(qid)
self.assertEqual(res.status_code, 200, res.text)
self.assertEqual(self.similar.call_args.args[2].id, 1)
self.login(self.mod)
self.assertEqual(self.chat(4).status_code, 200)
def test_tutor_course_attempt_modes_pool_and_review(self):
self.enroll()
self.assertEqual(self.chat(5).status_code, 403)
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)
self.db.get(Quiz, 2).allow_review = 0
self.db.commit()
self.assertEqual(self.chat(5, a.id).status_code, 403)
self.db.query(CourseEnrollment).delete()
self.db.commit()
self.assertEqual(self.chat(5, a.id).status_code, 403)
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.question_text = 'Long stem ' * 500
prompt = teach._build_system_prompt(q, [])
self.assertIn(q.question_text, prompt)
self.assertIn('8) Option 7', prompt)
self.assertIn(q.explanation, 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()))
self.assertIn('questions.user_id', sql)
self.assertIn('questions.is_shared', sql)
self.assertIn('course_id IS NOT 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)
for path in ('stem-1', 'stem-3', 'answer-1', 'answer-3'):
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')
self.assertEqual(self.client.get('/uploads/questions/stem-4.png').status_code, 404)
self.assertEqual(self.client.patch('/questions/3/share?shared=1').status_code, 401)
self.db.get(Question, 1).is_shared = 0
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_course_stems_answers_and_independent_references(self):
self.assertEqual(self.client.get('/uploads/questions/stem-5.png').status_code, 404)
self.enroll()
self.assertEqual(self.client.get('/uploads/questions/stem-5.png').status_code, 200)
self.assertEqual(self.client.get('/uploads/questions/answer-5.png').status_code, 404)
a = self.attempt(mode='exam', ids=[5])
url = f'/uploads/questions/answer-5.png?attempt_id={a.id}'
self.assertEqual(self.client.get(url).status_code, 404)
a.mode = 'study'
self.db.commit()
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.get(Quiz, 2).allow_review = 0
self.db.commit()
self.assertEqual(self.client.get(url).status_code, 404)
self.db.get(Question, 1).image_path = '/uploads/questions/answer-5.png'
self.db.commit()
self.assertEqual(self.client.get('/uploads/questions/answer-5.png').status_code, 200)
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()
self.enroll()
for field in ('image_path', 'explanation_image_path'):
for path in ('questions/stem-4.png', '/uploads/questions/answer-4.png', 'questions/stem-5.png', 'questions/2/draft.png', '../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')
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/answer-3.png', paths)
self.assertNotIn('questions/stem-4.png', paths)
self.assertNotIn('questions/stem-5.png', paths)
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_once()
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')
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).is_shared = 0
self.db.commit()
self.assertEqual(self.client.get('/uploads/legacy-public.svg').status_code, 404)
self.enroll()
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)
self.assertEqual(self.client.get(f'/flashcards/{deck.id}').status_code, 403)
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)
self.assertEqual(self.client.get('/uploads/legacy-private.svg').status_code, 404, 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)
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'<svg xmlns="http://www.w3.org/2000/svg"/>', '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'<html>Legacy SCORM</html>')
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()
Reference in a new issue View git blame Copy permalink