""" End-to-end + security test suite for Dr. Ideas Study Hub (92 checks). NEVER run against a production database: it creates users/courses and resets state. Setup (local only): 1. mariadb: create DB `drideas_test`, user dt/dtpass, import database/schema.sql 2. config/config.php pointing at that DB, 'env' => 'development' 3. php database/create_admin.php "Dr Ideas" admin@test.local 'AdminPass12345' 4. php -S 127.0.0.1:8099 -t public 5. pip install requests && python3 tests/e2e.py (run from the project root) Re-import schema.sql and recreate the admin before each re-run. """ import re, sys, io, subprocess import requests B = "http://127.0.0.1:8099" results = [] def check(name, cond, detail=""): results.append((name, bool(cond))) print(("PASS " if cond else "FAIL ") + name + ("" if cond else f" -> {detail}")) class NoFollow(requests.Session): def get(self, url, **kw): kw.setdefault("allow_redirects", False) return super().get(url, **kw) def request(self, method, url, **kw): kw.setdefault("allow_redirects", False) return super().request(method, url, **kw) def sess(): return NoFollow() def token(s, path): r = s.get(B + path, allow_redirects=True) m = re.search(r'name="csrf" value="([0-9a-f]+)"', r.text) return m.group(1) if m else None, r def post(s, path, data, tokpath, files=None, follow=False): t, _ = token(s, tokpath) data = dict(data); data["csrf"] = t return s.post(B + path, data=data, files=files, allow_redirects=follow) def sql(q): out = subprocess.run(["mariadb", "-N", "drideas_test", "-e", q], capture_output=True, text=True) return out.stdout.strip() # ---------- accounts ---------- admin, stu, stu2, anon = sess(), sess(), sess(), sess() r = post(admin, "/login.php", {"email": "admin@test.local", "password": "AdminPass12345"}, "/login.php") check("admin login redirects to /admin/", r.status_code == 302 and r.headers["Location"] == "/admin/", r.status_code) for s, name, email in [(stu, "Ama Mensah", "ama@test.local"), (stu2, "Kojo Asare", "kojo@test.local")]: r = post(s, "/register.php", {"name": name, "email": email, "student_id": "", "level": "100", "password": "StudentPass1", "password_confirm": "StudentPass1"}, "/register.php") check(f"register {name}", r.status_code == 302 and "dashboard" in r.headers["Location"], r.text[:200]) # ---------- access control ---------- for p in ["/admin/", "/admin/payments.php", "/admin/students.php", "/admin/analytics.php", "/admin/settings.php"]: r = stu.get(B + p) check(f"student blocked from {p}", r.status_code == 403, r.status_code) r = anon.get(B + p) check(f"anonymous redirected from {p}", r.status_code == 302 and "login" in r.headers["Location"], r.status_code) r = stu.get(B + "/admin/export.php?type=students") check("student cannot export CSV", r.status_code == 403, r.status_code) # ---------- CSRF ---------- r = stu.post(B + "/profile.php", data={"action": "details", "name": "Hacker", "level": "100"}) check("POST without CSRF token rejected (419)", r.status_code == 419, r.status_code) r = admin.post(B + "/admin/course_edit.php", data={"title": "x", "level": "100", "price": "1", "status": "draft"}) check("admin POST without CSRF rejected", r.status_code == 419, r.status_code) # ---------- admin builds a course ---------- r = post(admin, "/admin/settings.php", {"tutor_name": "Dr. Ideas", "payment_instructions": "MoMo 024 000 0000\nName: Dr Ideas", "payment_methods": "Mobile Money,Bank Transfer", "support_contact": ""}, "/admin/settings.php") check("settings saved", r.status_code == 302, r.text[:200]) r = post(admin, "/admin/course_edit.php", {"title": "Introduction to Biology", "description": "Cells and more", "level": "100", "price": "100.00", "cover_url": "", "status": "published", "access_days": ""}, "/admin/course_edit.php") check("course created", r.status_code == 302 and "curriculum" in r.headers["Location"], r.text[:300]) cid = int(sql("SELECT id FROM courses WHERE title='Introduction to Biology'")) cur = f"/admin/curriculum.php?course_id={cid}" post(admin, cur, {"action": "add_module", "title": "Cell Biology"}, cur) mid = int(sql(f"SELECT id FROM modules WHERE course_id={cid}")) post(admin, cur, {"action": "add_lesson", "module_id": mid, "title": "Cell Structure", "youtube": "https://youtu.be/dQw4w9WgXcQ"}, cur) post(admin, cur, {"action": "add_lesson", "module_id": mid, "title": "Cell Division", "youtube": "dQw4w9WgXcQ"}, cur) l1, l2 = [int(x) for x in sql(f"SELECT id FROM lessons WHERE module_id={mid} ORDER BY position").split()] check("two lessons created", l1 and l2) check("youtube url normalised to id", sql(f"SELECT youtube_id FROM lessons WHERE id={l1}") == "dQw4w9WgXcQ") post(admin, cur, {"action": "add_lesson", "module_id": mid, "title": "Bad", "youtube": "not a link"}, cur) check("invalid YouTube link rejected", sql(f"SELECT COUNT(*) FROM lessons WHERE title='Bad'") == "0") # PDF note upload (valid) and fake PDF (invalid) pdf = b"%PDF-1.4\n1 0 obj<>endobj\ntrailer<>\n%%EOF" t, _ = token(admin, cur) r = admin.post(B + cur, data={"csrf": t, "action": "upload_note", "lesson_id": l1}, files={"note": ("notes.pdf", pdf, "application/pdf")}, allow_redirects=False) check("PDF note uploaded", sql(f"SELECT resource_path IS NOT NULL FROM lessons WHERE id={l1}") == "1") t, _ = token(admin, cur) r = admin.post(B + cur, data={"csrf": t, "action": "upload_note", "lesson_id": l2}, files={"note": ("evil.pdf", b"", "application/pdf")}, allow_redirects=False) check("non-PDF disguised as .pdf rejected", sql(f"SELECT resource_path IS NULL FROM lessons WHERE id={l2}") == "1") # ---------- locked content before payment ---------- r = stu.get(B + f"/learn.php?course={cid}") check("unpaid student redirected from lesson player", r.status_code == 302 and "course.php" in r.headers["Location"], r.status_code) r = stu.get(B + f"/lesson_note.php?id={l1}") check("unpaid student cannot download notes", r.status_code == 404, r.status_code) r = anon.get(B + f"/lesson_note.php?id={l1}") check("anonymous cannot download notes", r.status_code in (302, 404), r.status_code) r = anon.get(B + f"/course.php?id={cid}") check("public syllabus has no video IDs", "dQw4w9WgXcQ" not in r.text and "Cell Structure" in r.text) r = post(stu, "/lesson_action.php", {"lesson_id": l1, "action": "complete"}, "/profile.php") check("unpaid student cannot mark lesson done", r.status_code == 403, r.status_code) r = post(stu, "/report.php", {"lesson_id": l1, "issue_type": "concept", "message": "hello there"}, "/profile.php") check("unpaid student cannot post a report", r.status_code == 403, r.status_code) # ---------- payment flow ---------- pay = {"course_id": cid, "method": "Mobile Money", "reference": "MP26ABC123", "amount": "100.00", "paid_on": __import__("datetime").date.today().isoformat()} r = post(stu, "/pay.php", pay, f"/course.php?id={cid}") check("payment submitted -> /payments.php", r.status_code == 302 and "payments" in r.headers["Location"], r.text[:200]) check("submission does NOT unlock the course", sql("SELECT COUNT(*) FROM enrollments") == "0") r = stu.get(B + f"/learn.php?course={cid}") check("still locked while pending", r.status_code == 302) r = post(stu2, "/pay.php", dict(pay, reference="mp26abc123"), f"/course.php?id={cid}", follow=True) check("duplicate reference (case-insensitive) rejected", "already been submitted" in r.text, r.text[-400:]) check("no second submission stored", sql("SELECT COUNT(*) FROM payment_submissions") == "1") r = post(stu, "/pay.php", dict(pay, reference="ANOTHER999"), f"/course.php?id={cid}", follow=True) check("second pending payment for same course blocked", sql("SELECT COUNT(*) FROM payment_submissions") == "1") # receipt upload by second student png = bytes.fromhex("89504e470d0a1a0a0000000d49484452000000010000000108060000001f15c4890000000d4944415478da63fcffff3f0300050001a5f645400000000049454e44ae426082") t, _ = token(stu2, f"/course.php?id={cid}") r = stu2.post(B + "/pay.php", data={"csrf": t, **dict(pay, reference="KOJO-REF-777")}, files={"receipt": ("r.png", png, "image/png")}, allow_redirects=False) pid2 = int(sql("SELECT id FROM payment_submissions WHERE reference='KOJO-REF-777'")) check("receipt stored for student 2", sql(f"SELECT receipt_path IS NOT NULL FROM payment_submissions WHERE id={pid2}") == "1") r = stu.get(B + f"/receipt.php?id={pid2}") check("student 1 cannot view student 2's receipt", r.status_code == 404, r.status_code) r = stu2.get(B + f"/receipt.php?id={pid2}") check("owner can view own receipt", r.status_code == 200 and r.headers["Content-Type"] == "image/png", r.status_code) r = admin.get(B + f"/receipt.php?id={pid2}") check("admin can view receipt", r.status_code == 200) r = anon.get(B + f"/receipt.php?id={pid2}") check("anonymous cannot view receipt", r.status_code == 302) t, _ = token(stu2, f"/course.php?id={cid}") r = stu2.post(B + "/pay.php", data={"csrf": t, **dict(pay, reference="TXT-EVIL-1")}, files={"receipt": ("x.png", b"", "image/png")}, allow_redirects=False) check("fake image receipt rejected", sql("SELECT COUNT(*) FROM payment_submissions WHERE reference='TXT-EVIL-1'") == "0") _name = sql("SELECT receipt_path FROM payment_submissions WHERE receipt_path IS NOT NULL LIMIT 1") _r = requests.get(B + "/storage/receipts/" + _name, allow_redirects=False) check("receipt file not directly web-reachable", not _r.headers.get("Content-Type", "").startswith("image/") and b"\x89PNG" not in _r.content, _r.headers.get("Content-Type")) _r = requests.get(B + "/../storage/receipts/" + _name, allow_redirects=False) check("path traversal to storage/ not served", b"\x89PNG" not in _r.content) # ---------- admin review ---------- pid1 = int(sql("SELECT id FROM payment_submissions WHERE reference='MP26ABC123'")) rv = f"/admin/payment_review.php?id={pid1}" r = post(admin, rv, {"decision": "approve"}, rv, follow=True) check("approve without verification tick refused", sql(f"SELECT status FROM payment_submissions WHERE id={pid1}") == "pending") r = post(admin, rv, {"decision": "reject", "reason": ""}, rv, follow=True) check("reject without reason refused", sql(f"SELECT status FROM payment_submissions WHERE id={pid1}") == "pending") rv2 = f"/admin/payment_review.php?id={pid2}" post(admin, rv2, {"decision": "reject", "reason": "Reference not in MoMo statement"}, rv2) check("payment rejected with reason", sql(f"SELECT status FROM payment_submissions WHERE id={pid2}") == "rejected") r = stu2.get(B + "/payments.php") check("student sees rejection reason", "Reference not in MoMo statement" in r.text) post(admin, rv, {"decision": "approve", "verified": "on"}, rv) check("payment approved", sql(f"SELECT status FROM payment_submissions WHERE id={pid1}") == "approved") check("enrolment created (source=payment, amount snapshot)", sql("SELECT CONCAT(source,'|',amount_paid,'|',status) FROM enrollments") == "payment|100.00|active") check("access grant logged", sql("SELECT COUNT(*) FROM access_grants WHERE action='grant'") == "1") check("approval audit-logged with admin id", sql("SELECT COUNT(*) FROM audit_logs WHERE action='payment.approve' AND user_id IS NOT NULL") == "1") post(admin, rv, {"decision": "reject", "reason": "changed my mind"}, rv) check("already-reviewed payment cannot be flipped", sql(f"SELECT status FROM payment_submissions WHERE id={pid1}") == "approved") # ---------- learning ---------- r = stu.get(B + f"/learn.php?course={cid}") check("enrolled student opens lesson player", r.status_code == 200 and "youtube-nocookie.com/embed/dQw4w9WgXcQ" in r.text, r.status_code) r = stu.get(B + f"/lesson_note.php?id={l1}") check("enrolled student downloads PDF note", r.status_code == 200 and r.headers["Content-Type"] == "application/pdf", r.status_code) r = stu2.get(B + f"/learn.php?course={cid}") check("student with REJECTED payment still locked out", r.status_code == 302) post(stu, "/lesson_action.php", {"lesson_id": l1, "action": "bookmark"}, f"/learn.php?course={cid}&lesson={l1}") check("bookmark does not count as completed", sql(f"SELECT COUNT(*) FROM lesson_progress WHERE user_id=(SELECT id FROM users WHERE email='ama@test.local') AND completed_at IS NOT NULL") == "0") r = stu.get(B + "/dashboard.php") check("dashboard: 0/2 topics after bookmark only", "0/2" in r.text) post(stu, "/lesson_action.php", {"lesson_id": l1, "action": "complete"}, f"/learn.php?course={cid}&lesson={l1}") r = stu.get(B + "/dashboard.php") check("dashboard: 1/2 and 50% after Mark as Done", "1/2" in r.text and "50%" in r.text) check("dashboard 'Next' points at lesson 2", "Next: Cell Division" in r.text, re.findall(r"Next: [^<]*", r.text)) post(stu, "/lesson_action.php", {"lesson_id": l1, "action": "uncomplete"}, f"/learn.php?course={cid}&lesson={l1}") r = stu.get(B + "/dashboard.php") check("undo completion returns to 0/2", "0/2" in r.text) post(stu, "/lesson_action.php", {"lesson_id": l1, "action": "complete"}, f"/learn.php?course={cid}&lesson={l1}") # resume: open lesson 2, then course entry should land on lesson 2 stu.get(B + f"/learn.php?course={cid}&lesson={l2}") r = stu.get(B + f"/learn.php?course={cid}") check("resume lands on last unfinished lesson", "Cell Division" in re.search(r"]*>(.*?)", r.text, re.S).group(1)) # reports + replies post(stu, "/report.php", {"lesson_id": l2, "issue_type": "concept", "message": "I do not get mitosis"}, f"/learn.php?course={cid}&lesson={l2}") rid = int(sql("SELECT id FROM topic_reports LIMIT 1")) r = admin.get(B + "/admin/reports.php", allow_redirects=True) check("admin inbox shows report + confusion count", "I do not get" not in r.text and "Cell Division" in r.text and "1 student" in r.text) post(admin, f"/admin/report.php?id={rid}", {"action": "reply", "message": "Think of it as copying then splitting.", "status": "resolved"}, f"/admin/report.php?id={rid}") r = stu.get(B + "/help.php") check("student sees tutor reply", "copying then splitting" in r.text and "Resolved" in r.text) r = stu2.get(B + "/help.php") check("other student cannot see it", "copying then splitting" not in r.text) # announcements post(admin, "/admin/announcements.php", {"action": "create", "course_id": cid, "title": "Quiz Friday", "body": "Revise cells"}, "/admin/announcements.php") check("enrolled student sees course announcement", "Quiz Friday" in stu.get(B + "/dashboard.php").text) check("non-enrolled student does not", "Quiz Friday" not in stu2.get(B + "/dashboard.php").text) # deletion safety post(admin, cur, {"action": "delete_lesson", "lesson_id": l1}, cur) check("lesson with student progress cannot be deleted", sql(f"SELECT COUNT(*) FROM lessons WHERE id={l1}") == "1") post(admin, cur, {"action": "delete_module", "module_id": mid}, cur) check("module with student progress cannot be deleted", sql(f"SELECT COUNT(*) FROM modules WHERE id={mid}") == "1") # IDOR: admin form can't touch another course's lesson post(admin, "/admin/course_edit.php", {"title": "Other", "description": "", "level": "200", "price": "0", "cover_url": "", "status": "draft", "access_days": ""}, "/admin/course_edit.php") cid2 = int(sql("SELECT id FROM courses WHERE title='Other'")) cur2 = f"/admin/curriculum.php?course_id={cid2}" post(admin, cur2, {"action": "update_lesson", "module_id": mid, "lesson_id": l1, "title": "HACKED", "youtube": "", "position": 1}, cur2) check("cross-course lesson edit blocked", sql(f"SELECT title FROM lessons WHERE id={l1}") == "Cell Structure") # draft course hidden r = anon.get(B + f"/course.php?id={cid2}") check("draft course 404 for the public", r.status_code == 404, r.status_code) r = stu.get(B + f"/learn.php?course={cid2}") check("student cannot open draft course player", r.status_code == 302) # ---------- manual / complimentary / revoke / extend / expiry ---------- uid2 = int(sql("SELECT id FROM users WHERE email='kojo@test.local'")) sp = f"/admin/student.php?id={uid2}" post(admin, sp, {"action": "grant", "course_id": cid, "source": "complimentary", "reason": ""}, sp) check("grant without reason refused", sql("SELECT COUNT(*) FROM enrollments") == "1") post(admin, sp, {"action": "grant", "course_id": cid, "source": "complimentary", "reason": "Scholarship"}, sp) check("complimentary grant created", sql(f"SELECT source FROM enrollments WHERE user_id={uid2}") == "complimentary") check("kojo can now learn", stu2.get(B + f"/learn.php?course={cid}").status_code == 200) eid = int(sql(f"SELECT id FROM enrollments WHERE user_id={uid2}")) post(admin, sp, {"action": "revoke", "enrollment_id": eid, "reason": "Test revoke"}, sp) check("revoked student locked out", stu2.get(B + f"/learn.php?course={cid}").status_code == 302) post(admin, sp, {"action": "grant", "course_id": cid, "source": "manual", "reason": "Paid cash"}, sp) check("re-grant after revoke works (no duplicate rows)", sql(f"SELECT COUNT(*) FROM enrollments WHERE user_id={uid2}") == "1" and sql(f"SELECT status FROM enrollments WHERE user_id={uid2}") == "active") # time-limited access sql(f"UPDATE courses SET access_days=30 WHERE id={cid}") sql(f"UPDATE enrollments SET expires_at = NOW() - INTERVAL 1 DAY WHERE user_id={uid2}") check("expired enrolment blocks access immediately (no cron needed)", stu2.get(B + f"/learn.php?course={cid}").status_code == 302) out = subprocess.run(["php", "cron/expire_access.php"], capture_output=True, text=True, cwd="/home/claude/drideas") check("cron marks it expired", "expired=1" in out.stdout and sql(f"SELECT status FROM enrollments WHERE user_id={uid2}") == "expired", out.stdout + out.stderr) post(admin, sp, {"action": "extend", "enrollment_id": eid, "days": 30, "reason": "Late payment grace"}, sp) check("extend re-activates and restores access", stu2.get(B + f"/learn.php?course={cid}").status_code == 200) # ---------- analytics + CSV ---------- r = admin.get(B + "/admin/analytics.php") check("analytics page renders with revenue", r.status_code == 200 and "GHS 100.00" in r.text, r.status_code) sql("UPDATE users SET name='=HYPERLINK(\"http://evil\",\"x\")' WHERE email='kojo@test.local'") r = admin.get(B + "/admin/export.php?type=students") check("students CSV downloads", r.status_code == 200 and "text/csv" in r.headers["Content-Type"]) check("CSV neutralises formula injection", "'=HYPERLINK" in r.text and ",=HYPERLINK" not in r.text and '"=HYPERLINK' not in r.text, r.text[:400]) for t in ["enrollments", "payments"]: r = admin.get(B + f"/admin/export.php?type={t}") check(f"{t} CSV downloads", r.status_code == 200 and r.text.count("\n") >= 2, r.status_code) check("exports are audit-logged", int(sql("SELECT COUNT(*) FROM audit_logs WHERE action LIKE 'export.%'")) >= 3) # ---------- auth hardening ---------- bad = sess() codes = [] for i in range(6): r = post(bad, "/login.php", {"email": "ama@test.local", "password": "wrongwrong"}, "/login.php", follow=True) codes.append("Too many failed attempts" in r.text) check("login rate-limited after 5 failures", codes[-1]) good = sess() r = post(good, "/login.php", {"email": "ama@test.local", "password": "StudentPass1"}, "/login.php", follow=True) check("correct password also blocked during lockout (same IP+email)", "Too many failed attempts" in r.text) sql("DELETE FROM login_attempts") # profile password change r = post(stu, "/profile.php", {"action": "password", "current_password": "nope", "new_password": "NewStudentPass2", "confirm_password": "NewStudentPass2"}, "/profile.php", follow=True) check("password change needs correct current password", "current password is incorrect" in r.text) post(stu, "/profile.php", {"action": "password", "current_password": "StudentPass1", "new_password": "NewStudentPass2", "confirm_password": "NewStudentPass2"}, "/profile.php") fresh = sess() r = post(fresh, "/login.php", {"email": "ama@test.local", "password": "NewStudentPass2"}, "/login.php") check("can log in with new password", r.status_code == 302 and "dashboard" in r.headers["Location"]) # XSS escaping post(stu, "/profile.php", {"action": "details", "name": "", "student_id": "", "level": "100"}, "/profile.php") r = admin.get(B + "/admin/students.php") check("names are HTML-escaped", "" not in r.text and "<script>" in r.text) # SQLi probe on search r = admin.get(B + "/admin/students.php", params={"q": "' OR 1=1 -- "}) check("SQL injection probe returns no rows / no error", r.status_code == 200 and "No students found" in r.text) # headers / dotfiles r = anon.get(B + "/") check("security headers present", r.headers.get("X-Frame-Options") == "SAMEORIGIN" and r.headers.get("X-Content-Type-Options") == "nosniff") failed = [n for n, ok in results if not ok] print(f"\n{len(results) - len(failed)}/{len(results)} checks passed") if failed: print("FAILED:", *failed, sep="\n - ") sys.exit(1)