`GET /api/contact/submissions` had no authentication. `require_admin` was imported inside the function body and never used as a dependency, so the import read as protection and was none: anyone who guessed the path could read every sender's name, email address and message. `PUT .../read` was open the same way. Both now depend on `require_admin`, with a test that a learner gets 403 and an administrator gets the list. A row with a null timestamp no longer takes the whole listing down with it — which is the only reason the hole showed up as a 500 rather than as data. Also: the tutor's site switch lives in Redis, which the tests share with the running site, so turning the tutor off in the interface turned a test red. The test now sets the flag it depends on and puts it back. And the tutor button is hidden until the server says it is allowed, rather than shown and then withdrawn — on a site with it switched off that flicker reads as a bug rather than a policy. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01TqXevQJhxFrM7jJg82cgZN
78 lines
3.2 KiB
Python
78 lines
3.2 KiB
Python
"""Who may read the contact form's messages.
|
|
|
|
Run: DATABASE_URL=sqlite:// PYTHONPATH=backend python -m unittest discover -s backend/tests
|
|
"""
|
|
import os
|
|
os.environ.setdefault("DATABASE_URL", "sqlite://")
|
|
|
|
import unittest
|
|
|
|
from fastapi import FastAPI
|
|
from fastapi.testclient import TestClient
|
|
from sqlalchemy import create_engine, text
|
|
from sqlalchemy.orm import Session
|
|
from sqlalchemy.pool import StaticPool
|
|
|
|
from app.database import get_db
|
|
from app.models.user import User
|
|
from app.routers import contact
|
|
from app.utils.auth import get_current_user
|
|
|
|
|
|
class ContactPrivacyTests(unittest.TestCase):
|
|
"""The listing was open to the internet.
|
|
|
|
`require_admin` was imported inside the function body and never used as a
|
|
dependency, so the import read as protection and was none: anyone who
|
|
guessed the path could read every sender's name, address and message.
|
|
"""
|
|
|
|
def setUp(self):
|
|
self.engine = create_engine("sqlite://", connect_args={"check_same_thread": False},
|
|
poolclass=StaticPool)
|
|
# Only the table this router touches, plus users. `create_all` pulls in
|
|
# every model in the metadata, and one of them carries a raw-SQL table
|
|
# SQLite cannot build.
|
|
User.__table__.create(self.engine, checkfirst=True)
|
|
self.db = Session(self.engine)
|
|
self.db.execute(text(
|
|
"CREATE TABLE IF NOT EXISTS contact_submissions ("
|
|
"id INTEGER PRIMARY KEY, name TEXT, email TEXT, type TEXT, message TEXT,"
|
|
" read INTEGER DEFAULT 0, created_at TIMESTAMP)"))
|
|
self.db.execute(text(
|
|
"INSERT INTO contact_submissions (id, name, email, type, message, read, created_at)"
|
|
" VALUES (1, 'A Sender', 'sender@example.test', 'question', 'Private message', 0, NULL)"))
|
|
self.db.commit()
|
|
|
|
self.admin = User(id=1, email="admin@example.test", name="Admin", role="admin")
|
|
self.learner = User(id=2, email="learner@example.test", name="Learner", role="user")
|
|
self.db.add_all([self.admin, self.learner])
|
|
self.db.commit()
|
|
|
|
self.user = self.learner
|
|
app = FastAPI()
|
|
app.include_router(contact.router, prefix="/contact")
|
|
app.dependency_overrides[get_db] = lambda: self.db
|
|
app.dependency_overrides[get_current_user] = lambda: self.user
|
|
self.client = TestClient(app)
|
|
|
|
def tearDown(self):
|
|
self.db.close()
|
|
self.engine.dispose()
|
|
|
|
def test_a_learner_may_not_read_or_touch_the_messages(self):
|
|
self.assertEqual(self.client.get("/contact/submissions").status_code, 403)
|
|
self.assertEqual(self.client.put("/contact/submissions/1/read").status_code, 403)
|
|
|
|
def test_an_administrator_may(self):
|
|
self.user = self.admin
|
|
listing = self.client.get("/contact/submissions")
|
|
self.assertEqual(listing.status_code, 200, listing.text)
|
|
self.assertEqual(listing.json()[0]["email"], "sender@example.test")
|
|
# A row with no timestamp does not take the whole listing down with it.
|
|
self.assertIsNone(listing.json()[0]["created_at"])
|
|
self.assertEqual(self.client.put("/contact/submissions/1/read").status_code, 200)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|