625 lines
26 KiB
Python
625 lines
26 KiB
Python
import asyncio
|
|
import sqlite3
|
|
import threading
|
|
import tempfile
|
|
import time
|
|
import smtplib
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
import feedback_service
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
from feedback_service import Settings, SendFailure, UncertainFailure, SmtpSender, SubmissionStore, create_app
|
|
|
|
|
|
ID = "cbaaaace-60b2-4871-8a95-ae316444a2fb"
|
|
PNG = b"\x89PNG\r\n\x1a\n" + b"safe fixture"
|
|
HTML = (b'<!doctype html><!-- TPSchedule school diagnostic --><html><head><meta charset="utf-8">'
|
|
b'<meta http-equiv="Content-Security-Policy" content="default-src \'none\'; base-uri \'none\'; form-action \'none\'">'
|
|
b'</head><body><table></table></body></html>')
|
|
|
|
|
|
def test_real_client_generated_synthetic_html_is_accepted_as_attachment(service):
|
|
client, sender, _ = service
|
|
page = (Path(__file__).parent / 'fixtures/client-school-diagnostic.html').read_bytes()
|
|
fields = [('submission_id', (None, ID)), ('type', (None, 'problem')), ('message', (None, 'Cannot import'))]
|
|
response = client.post('/v1/feedback', files=fields + [('diagnostic_html', ('page.html', page, 'text/html'))])
|
|
assert response.status_code == 200
|
|
attachment = list(sender.messages[0].iter_attachments())[0]
|
|
assert attachment.get_payload(decode=True) == page
|
|
|
|
|
|
@pytest.mark.parametrize('fixture', ['client-graduate-timetable.html', 'client-graduate-exams.html'])
|
|
def test_real_graduate_client_html_is_mailed_without_changing_its_evidence(service, fixture):
|
|
client, sender, _ = service
|
|
page = (Path(__file__).parent / 'fixtures' / fixture).read_bytes()
|
|
fields = [('submission_id', (None, ID)), ('type', (None, 'problem')), ('message', (None, 'Graduate import failed'))]
|
|
response = client.post('/v1/feedback', files=fields + [('diagnostic_html', ('page.html', page, 'text/html'))])
|
|
assert response.status_code == 200
|
|
attachment = list(sender.messages[0].iter_attachments())[0]
|
|
assert attachment.get_filename() == 'TPSchedule-school-page.html'
|
|
assert attachment.get_payload(decode=True) == page
|
|
|
|
|
|
@pytest.mark.parametrize('extra', [
|
|
'<select id="account"><option value="73">Account</option></select>',
|
|
'<select id="drpxq" onchange="steal()"><option value="73">Term</option></select>',
|
|
'<select id="xqbh"><option value="https://evil.test/">Term</option></select>',
|
|
'<option value="73" selected>Outside semester selector</option>',
|
|
'<select id="drpxq"><option value="73" selected onclick="steal()">Term</option></select>',
|
|
'<table id="account"><tr><td id="token">Secret</td></tr></table>',
|
|
])
|
|
def test_graduate_diagnostics_reject_unrelated_or_active_controls(service, extra):
|
|
client, sender, _ = service
|
|
page = HTML.replace(b'</body>', extra.encode() + b'</body>')
|
|
fields = [('submission_id', (None, ID)), ('type', (None, 'problem')), ('message', (None, 'Graduate import failed'))]
|
|
response = client.post('/v1/feedback', files=fields + [('diagnostic_html', ('page.html', page, 'text/html'))])
|
|
assert response.status_code == 415
|
|
assert not sender.messages
|
|
|
|
|
|
@pytest.mark.parametrize('extra', [
|
|
'<script>alert(1)</script>', '<img src="https://evil.test/">',
|
|
'<table onclick="alert(1)"></table>', '<meta http-equiv="refresh" content="0;url=https://evil.test/">',
|
|
'<iframe srcdoc="x"></iframe>',
|
|
])
|
|
def test_html_diagnostic_rejects_active_or_network_content_even_with_marker(service, extra):
|
|
client, sender, _ = service
|
|
data = HTML.replace(b'</body>', extra.encode() + b'</body>')
|
|
fields = [('submission_id', (None, ID)), ('type', (None, 'problem')), ('message', (None, 'Cannot import'))]
|
|
response = client.post('/v1/feedback', files=fields + [('diagnostic_html', ('page.html', data, 'text/html'))])
|
|
assert response.status_code == 415
|
|
assert not sender.messages
|
|
|
|
|
|
def test_html_diagnostic_is_fifth_attachment_and_retry_uses_identical_snapshot(service):
|
|
client, sender, _ = service
|
|
files = [('submission_id', (None, ID)), ('type', (None, 'problem')), ('message', (None, 'Cannot import'))]
|
|
files += [('attachments', (f'{i}.png', PNG, 'image/png')) for i in range(3)]
|
|
files += [('diagnostics', ('log.txt', b'TPSchedule diagnostic log\n', 'text/plain')),
|
|
('diagnostic_html', ('page.html', HTML, 'text/html; charset=utf-8'))]
|
|
assert client.post('/v1/feedback', files=files).status_code == 200
|
|
attachments = list(sender.messages[0].iter_attachments())
|
|
assert len(attachments) == 5
|
|
assert attachments[-1].get_filename() == 'TPSchedule-school-page.html'
|
|
assert attachments[-1].get_payload(decode=True) == HTML
|
|
assert client.post('/v1/feedback', files=files).status_code == 200
|
|
assert len(sender.messages) == 1
|
|
files[-1] = ('diagnostic_html', ('page.html', HTML.replace(b'<table>', b'<table><tr><td>changed</td></tr>'), 'text/html'))
|
|
assert client.post('/v1/feedback', files=files).status_code == 409
|
|
|
|
|
|
@pytest.mark.parametrize('data,mime,kind,status', [
|
|
(b'not HTML', 'text/html', 'problem', 415),
|
|
(HTML, 'image/png', 'problem', 415),
|
|
(HTML + b'\xff', 'text/html', 'problem', 415),
|
|
(HTML + b'x' * (2 * 1024 * 1024), 'text/html', 'problem', 413),
|
|
(HTML, 'text/html', 'development', 422),
|
|
], ids=['header', 'mime', 'utf8', 'oversize', 'development'])
|
|
def test_invalid_html_diagnostics_never_send(service, data, mime, kind, status):
|
|
client, sender, _ = service
|
|
fields = [('submission_id', (None, ID)), ('type', (None, kind))]
|
|
fields += [('message', (None, 'Cannot import'))] if kind == 'problem' else [('school', (None, 'School')), ('contact', (None, 'Contact'))]
|
|
response = client.post('/v1/feedback', files=fields + [('diagnostic_html', ('page.html', data, mime))])
|
|
assert response.status_code == status
|
|
assert not sender.messages
|
|
|
|
|
|
class RecordingSender:
|
|
def __init__(self):
|
|
self.messages = []
|
|
self.failure = None
|
|
|
|
def __call__(self, message):
|
|
if self.failure:
|
|
raise self.failure
|
|
self.messages.append(message)
|
|
|
|
|
|
@pytest.fixture
|
|
def service(tmp_path, monkeypatch):
|
|
monkeypatch.setattr(tempfile, "tempdir", str(tmp_path))
|
|
settings = Settings(database_path=str(tmp_path / "feedback.sqlite3"),
|
|
smtp_username="feedback@example.com", smtp_password="test-only",
|
|
mail_from="feedback@example.com", mail_to="owner@example.com",
|
|
idempotency_secret="test-key-at-least-thirty-two-characters")
|
|
sender = RecordingSender()
|
|
app = create_app(settings, sender)
|
|
with TestClient(app) as client:
|
|
yield client, sender, settings
|
|
|
|
|
|
def post(client, **overrides):
|
|
fields = {"submission_id": ID, "type": "problem", "message": "The timetable is missing."}
|
|
fields.update(overrides)
|
|
return client.post("/v1/feedback", files=[(k, (None, v)) for k, v in fields.items()])
|
|
|
|
|
|
def test_success_only_after_sender_accepted_and_retry_is_durable(service):
|
|
client, sender, settings = service
|
|
response = post(client, contact="student@example.com")
|
|
assert response.status_code == 200
|
|
assert response.json() == {"success": True, "submission_id": ID, "status": "sent"}
|
|
assert len(sender.messages) == 1
|
|
assert sender.messages[0]["From"] == "feedback@example.com"
|
|
assert sender.messages[0]["To"] == "owner@example.com"
|
|
assert "student@example.com" in sender.messages[0].get_body().get_content()
|
|
# A new application instance reads the persisted status, not an in-memory cache.
|
|
with TestClient(create_app(settings, sender)) as restarted:
|
|
assert post(restarted, contact="student@example.com").json()["status"] == "sent"
|
|
assert len(sender.messages) == 1
|
|
|
|
|
|
def test_diagnostic_log_is_extra_email_attachment_and_retry_stays_idempotent(service):
|
|
client, sender, _ = service
|
|
log = b'TPSchedule diagnostic log\n{"event":"login_opened"}\n'
|
|
files = [("submission_id", (None, ID)), ("type", (None, "problem")),
|
|
("message", (None, "Login stopped"))] + [
|
|
("attachments", (f"image-{i}.png", PNG, "image/png")) for i in range(3)] + [
|
|
("diagnostics", ("TPSchedule-diagnostics.txt", log, "text/plain; charset=utf-8"))]
|
|
assert client.post('/v1/feedback', files=files).status_code == 200
|
|
attachments = list(sender.messages[0].iter_attachments())
|
|
assert len(attachments) == 4
|
|
assert attachments[-1].get_filename() == 'TPSchedule-diagnostics.txt'
|
|
assert attachments[-1].get_payload(decode=True) == log
|
|
assert client.post('/v1/feedback', files=files).status_code == 200
|
|
assert len(sender.messages) == 1
|
|
|
|
|
|
@pytest.mark.parametrize('name,kind,data,mime,status', [
|
|
('diagnostics', 'problem', b'not a log', 'text/plain', 415),
|
|
('diagnostics', 'problem', b'TPSchedule diagnostic log\n\xff', 'text/plain', 415),
|
|
('diagnostics', 'problem', b'TPSchedule diagnostic log\n' + b'x' * (256 * 1024), 'text/plain', 413),
|
|
('diagnostics', 'development', b'TPSchedule diagnostic log\n', 'text/plain', 422),
|
|
], ids=['header', 'utf8', 'oversize', 'development'])
|
|
def test_invalid_diagnostics_never_send(service, name, kind, data, mime, status):
|
|
client, sender, _ = service
|
|
fields = [('submission_id', (None, ID)), ('type', (None, kind))]
|
|
if kind == 'problem':
|
|
fields += [('message', (None, 'Cannot import'))]
|
|
else:
|
|
fields += [('school', (None, 'University')), ('contact', (None, 'QQ123'))]
|
|
response = client.post('/v1/feedback', files=fields + [(name, ('log.txt', data, mime))])
|
|
assert response.status_code == status
|
|
assert not sender.messages
|
|
|
|
|
|
def test_reusing_id_with_other_content_rejected(service):
|
|
client, sender, _ = service
|
|
assert post(client).status_code == 200
|
|
assert post(client, message="Changed complaint").status_code == 409
|
|
assert len(sender.messages) == 1
|
|
|
|
|
|
@pytest.mark.parametrize("fields", [
|
|
{"message": ""}, {"message": "x" * 5001}, {"contact": "x" * 201},
|
|
{"school": "unexpected"}, {"type": "other"}, {"submission_id": "too-short"},
|
|
{"type": "development", "school": "Example school", "contact": ""},
|
|
{"type": "development", "school": "", "contact": "12345678"},
|
|
{"type": "development", "school": "Example school", "contact": "12345678"},
|
|
])
|
|
def test_invalid_fields_do_not_send(service, fields):
|
|
client, sender, _ = service
|
|
assert post(client, **fields).status_code == 422
|
|
assert not sender.messages
|
|
|
|
|
|
def test_development_has_only_school_and_contact(service):
|
|
client, sender, _ = service
|
|
response = client.post("/v1/feedback", files=[
|
|
("submission_id", (None, ID)), ("type", (None, "development")),
|
|
("school", (None, "Example school")), ("contact", (None, "student@example.com")),
|
|
])
|
|
assert response.status_code == 200
|
|
body = sender.messages[0].get_body().get_content()
|
|
assert "Example school" in body and "student@example.com" in body
|
|
|
|
|
|
def test_attachment_is_sanitized_and_type_checked(service):
|
|
client, sender, _ = service
|
|
files = [("submission_id", (None, ID)), ("type", (None, "problem")),
|
|
("message", (None, "Attachment")),
|
|
("attachments", ("../../bad\r\nname.png", PNG, "image/png"))]
|
|
response = client.post("/v1/feedback", files=files)
|
|
assert response.status_code == 200
|
|
attachment = list(sender.messages[0].iter_attachments())[0]
|
|
assert attachment.get_filename() == "bad_0D_0Aname.png"
|
|
assert attachment.get_payload(decode=True) == PNG
|
|
files[-1] = ("attachments", ("pretend.png", b"<script>alert(1)</script>", "image/png"))
|
|
assert client.post("/v1/feedback", files=files).status_code == 415
|
|
|
|
|
|
def test_development_rejects_media(service):
|
|
client, sender, _ = service
|
|
response = client.post("/v1/feedback", files=[
|
|
("submission_id", (None, ID)), ("type", (None, "development")),
|
|
("school", (None, "Example")), ("contact", (None, "student@example.com")),
|
|
("attachments", ("a.png", PNG, "image/png")),
|
|
])
|
|
assert response.status_code == 422
|
|
assert not sender.messages
|
|
|
|
|
|
def test_duplicate_fields_and_unexpected_fields_rejected(service):
|
|
client, sender, _ = service
|
|
for extra in [("message", (None, "second")), ("recipient", (None, "victim@example.com"))]:
|
|
response = client.post("/v1/feedback", files=[
|
|
("submission_id", (None, ID)), ("type", (None, "problem")),
|
|
("message", (None, "Complaint")), extra,
|
|
])
|
|
assert response.status_code == 422
|
|
assert not sender.messages
|
|
|
|
|
|
def test_definite_smtp_failure_is_non_success_and_can_retry(service):
|
|
client, sender, _ = service
|
|
sender.failure = SendFailure()
|
|
response = post(client)
|
|
assert response.status_code == 503
|
|
assert response.json()["success"] is False
|
|
assert response.json()["status"] == "retryable"
|
|
sender.failure = None
|
|
assert post(client).json()["status"] == "sent"
|
|
|
|
|
|
def test_uncertain_smtp_acceptance_never_auto_resends(service):
|
|
client, sender, settings = service
|
|
sender.failure = UncertainFailure()
|
|
response = post(client)
|
|
assert response.status_code == 503
|
|
assert response.json()["status"] == "uncertain"
|
|
sender.failure = None
|
|
with TestClient(create_app(settings, sender)) as restarted:
|
|
response = post(restarted)
|
|
assert response.status_code == 503
|
|
assert response.json()["error"] == "delivery_uncertain"
|
|
assert not sender.messages
|
|
|
|
|
|
def test_parallel_retry_does_not_send_twice(service):
|
|
client, sender, _ = service
|
|
started, finish = threading.Event(), threading.Event()
|
|
|
|
def blocking_sender(message):
|
|
started.set()
|
|
assert finish.wait(10)
|
|
sender(message)
|
|
|
|
client.app.state.sender = blocking_sender
|
|
result = []
|
|
worker = threading.Thread(target=lambda: result.append(post(client)))
|
|
worker.start()
|
|
assert started.wait(10)
|
|
try:
|
|
response = post(client)
|
|
assert response.status_code == 409
|
|
assert response.json()["status"] == "in_progress"
|
|
finally:
|
|
finish.set()
|
|
worker.join(10)
|
|
assert result[0].status_code == 200
|
|
assert len(sender.messages) == 1
|
|
|
|
|
|
def test_stale_inflight_record_is_uncertain(service):
|
|
client, sender, settings = service
|
|
assert post(client).status_code == 200
|
|
with sqlite3.connect(settings.database_path) as db:
|
|
db.execute("UPDATE submissions SET status='sending', updated=?", (time.time() - 600,))
|
|
response = post(client)
|
|
assert response.status_code == 503
|
|
assert response.json()["status"] == "uncertain"
|
|
assert len(sender.messages) == 1
|
|
|
|
|
|
def test_file_count_and_individual_size_are_limited(service):
|
|
client, sender, _ = service
|
|
base = [("submission_id", (None, ID)), ("type", (None, "problem")),
|
|
("message", (None, "Complaint"))]
|
|
response = client.post("/v1/feedback", files=base + [
|
|
("attachments", (f"{i}.png", PNG, "image/png")) for i in range(4)])
|
|
assert response.status_code == 413
|
|
response = client.post("/v1/feedback", files=base + [
|
|
("attachments", ("large.png", PNG + b"x" * (8 * 1024 * 1024), "image/png"))])
|
|
assert response.status_code == 413
|
|
assert not sender.messages
|
|
|
|
|
|
def test_aggregate_attachments_are_limited(service):
|
|
client, sender, _ = service
|
|
base = [("submission_id", (None, ID)), ("type", (None, "problem")),
|
|
("message", (None, "Complaint"))]
|
|
response = client.post("/v1/feedback", files=base + [
|
|
("attachments", (f"{i}.png", PNG + b"x" * (7 * 1024 * 1024), "image/png"))
|
|
for i in range(3)])
|
|
assert response.status_code == 413
|
|
assert not sender.messages
|
|
|
|
|
|
def test_chunked_body_without_content_length_is_limited(service):
|
|
client, sender, _ = service
|
|
events, output = iter([
|
|
{"type": "http.request", "body": b"x" * (12 * 1024 * 1024), "more_body": True},
|
|
{"type": "http.request", "body": b"x" * (12 * 1024 * 1024), "more_body": False},
|
|
]), []
|
|
|
|
async def receive():
|
|
return next(events)
|
|
|
|
async def send(event):
|
|
output.append(event)
|
|
|
|
scope = {"type": "http", "asgi": {"version": "3.0"}, "method": "POST",
|
|
"path": "/v1/feedback", "raw_path": b"/v1/feedback", "query_string": b"",
|
|
"headers": [(b"content-type", b"multipart/form-data; boundary=abc")],
|
|
"client": ("127.0.0.1", 123), "server": ("localhost", 80), "scheme": "http"}
|
|
asyncio.run(client.app(scope, receive, send))
|
|
assert output[0]["status"] == 413
|
|
assert not sender.messages
|
|
|
|
|
|
def test_oversized_declared_body_is_rejected_before_parsing(service):
|
|
client, sender, _ = service
|
|
response = client.post("/v1/feedback", content=b"", headers={
|
|
"content-type": "multipart/form-data; boundary=abc", "content-length": str(100 * 1024 * 1024)})
|
|
assert response.status_code == 413
|
|
assert not sender.messages
|
|
|
|
|
|
def test_request_rate_is_limited_before_smtp(service):
|
|
_, sender, settings = service
|
|
settings.ip_requests_per_hour = 2
|
|
with TestClient(create_app(settings, sender)) as client:
|
|
assert post(client).status_code == 200
|
|
assert post(client).status_code == 200
|
|
response = post(client)
|
|
assert response.status_code == 429
|
|
assert response.headers["retry-after"]
|
|
assert len(sender.messages) == 1
|
|
|
|
|
|
def test_unsupported_body_format_cannot_use_unbounded_form_parser(service):
|
|
client, sender, _ = service
|
|
assert client.post("/v1/feedback", data={"type": "problem"}).status_code == 415
|
|
assert not sender.messages
|
|
|
|
|
|
def test_persistent_metadata_does_not_contain_personal_text(service):
|
|
client, _, settings = service
|
|
assert post(client, message="private complaint", contact="private@example.com").status_code == 200
|
|
with sqlite3.connect(settings.database_path) as db:
|
|
values = repr(db.execute("SELECT * FROM submissions").fetchall())
|
|
assert "private complaint" not in values
|
|
assert "private@example.com" not in values
|
|
|
|
|
|
def test_invalid_environment_addresses_fail_before_startup(tmp_path):
|
|
for sender in ["feedback@example.com\r\nBcc: victim@example.com", "Display Name <a@example.com>", ""]:
|
|
settings = Settings(database_path=str(tmp_path / "db"), smtp_username="user", smtp_password="secret",
|
|
mail_from=sender, mail_to="owner@example.com")
|
|
with pytest.raises(ValueError):
|
|
create_app(settings, RecordingSender())
|
|
|
|
|
|
def test_global_rate_cannot_be_bypassed_by_changing_client_ips(service):
|
|
_, sender, settings = service
|
|
settings.global_requests_per_hour = 2
|
|
app = create_app(settings, sender)
|
|
for ip, expected in [("1.2.3.4", 200), ("2.3.4.5", 200), ("3.4.5.6", 429)]:
|
|
with TestClient(app, client=(ip, 123)) as client:
|
|
assert post(client).status_code == expected
|
|
assert len(sender.messages) == 1
|
|
|
|
|
|
def test_concurrency_is_bounded_before_parsing(service):
|
|
_, sender, settings = service
|
|
settings.max_concurrent_requests = 1
|
|
started, finish = threading.Event(), threading.Event()
|
|
|
|
def blocking_sender(message):
|
|
started.set()
|
|
assert finish.wait(10)
|
|
sender(message)
|
|
|
|
with TestClient(create_app(settings, blocking_sender)) as client:
|
|
result = []
|
|
worker = threading.Thread(target=lambda: result.append(post(client)))
|
|
worker.start()
|
|
assert started.wait(10)
|
|
try:
|
|
response = post(client)
|
|
assert response.status_code == 503
|
|
assert response.json()["error"] == "server_busy"
|
|
finally:
|
|
finish.set()
|
|
worker.join(10)
|
|
assert result[0].status_code == 200
|
|
assert len(sender.messages) == 1
|
|
|
|
|
|
def test_lost_database_commit_after_acceptance_never_claims_success(service, monkeypatch):
|
|
client, sender, _ = service
|
|
|
|
def storage_failure(self, submission_id, status):
|
|
raise sqlite3.OperationalError("simulated storage failure")
|
|
|
|
monkeypatch.setattr(SubmissionStore, "finish", storage_failure)
|
|
response = post(client)
|
|
assert response.status_code == 503
|
|
assert response.json()["success"] is False
|
|
assert response.json()["error"] == "storage_unavailable"
|
|
assert post(client).status_code == 409
|
|
assert len(sender.messages) == 1
|
|
|
|
|
|
class ScriptedSmtp:
|
|
"""Protocol boundary fake; no socket is opened and no real mail is sent."""
|
|
def __init__(self, phase="accept"):
|
|
self.phase, self.events, self.context = phase, [], None
|
|
|
|
def login(self, user, password):
|
|
self.events.append("login")
|
|
if self.phase == "auth_fail":
|
|
raise smtplib.SMTPAuthenticationError(535, b"authentication rejected")
|
|
|
|
def ehlo(self):
|
|
self.events.append("ehlo")
|
|
|
|
def starttls(self, context):
|
|
self.context = context
|
|
self.events.append("starttls")
|
|
|
|
def mail(self, sender):
|
|
self.events.append(("mail", sender))
|
|
return 250, b"ok"
|
|
|
|
def rcpt(self, recipient):
|
|
self.events.append(("rcpt", recipient))
|
|
return (550, b"rejected") if self.phase == "recipient_fail" else (250, b"ok")
|
|
|
|
def data(self, payload):
|
|
self.events.append("data")
|
|
if self.phase == "lost_ack":
|
|
raise smtplib.SMTPServerDisconnected("connection lost")
|
|
if self.phase == "data_reject":
|
|
raise smtplib.SMTPDataError(552, b"too big")
|
|
return 250, b"accepted"
|
|
|
|
def quit(self):
|
|
self.events.append("quit")
|
|
if self.phase == "quit_fail":
|
|
raise smtplib.SMTPServerDisconnected("quit lost")
|
|
|
|
def close(self):
|
|
self.events.append("close")
|
|
|
|
|
|
@pytest.mark.parametrize("phase, expected", [
|
|
("accept", None), ("quit_fail", None), ("auth_fail", SendFailure),
|
|
("recipient_fail", SendFailure), ("data_reject", SendFailure), ("lost_ack", UncertainFailure),
|
|
])
|
|
def test_smtp_outcome_tracks_data_acknowledgement(service, monkeypatch, phase, expected):
|
|
client, sender, settings = service
|
|
assert post(client).status_code == 200
|
|
fake = ScriptedSmtp(phase)
|
|
|
|
def connect(host, port, timeout, context):
|
|
assert host == "smtp.feishu.cn" and port == 465
|
|
assert context.check_hostname and context.verify_mode == 2
|
|
return fake
|
|
|
|
monkeypatch.setattr(smtplib, "SMTP_SSL", connect)
|
|
if expected:
|
|
with pytest.raises(expected):
|
|
SmtpSender(settings)(sender.messages[0])
|
|
else:
|
|
SmtpSender(settings)(sender.messages[0])
|
|
assert ("mail", "feedback@example.com") in fake.events or phase == "auth_fail"
|
|
assert ("rcpt", "owner@example.com") in fake.events or phase == "auth_fail"
|
|
|
|
|
|
def test_starttls_is_verified_before_authentication(service, monkeypatch):
|
|
client, sender, settings = service
|
|
assert post(client).status_code == 200
|
|
settings.smtp_port, settings.smtp_tls = 587, "starttls"
|
|
fake = ScriptedSmtp()
|
|
monkeypatch.setattr(smtplib, "SMTP", lambda host, port, timeout: fake)
|
|
SmtpSender(settings)(sender.messages[0])
|
|
assert fake.events[:4] == ["ehlo", "starttls", "ehlo", "login"]
|
|
assert fake.context.check_hostname and fake.context.verify_mode == 2
|
|
|
|
|
|
@pytest.mark.parametrize("name, mime, data", [
|
|
("a.jpg", "image/jpeg", b"\xff\xd8\xff\xe0fixture"),
|
|
("a.gif", "image/gif", b"GIF89afixture"),
|
|
("a.webp", "image/webp", b"RIFF\x00\x00\x00\x00WEBPfixture"),
|
|
("a.heic", "image/heic", b"\x00\x00\x00\x14ftypheic\x00\x00\x00\x00mif1"),
|
|
("a.mp4", "video/mp4", b"\x00\x00\x00\x14ftypisom\x00\x00\x00\x00mp42"),
|
|
("a.m4v", "video/x-m4v", b"\x00\x00\x00\x14ftypM4V \x00\x00\x00\x00mp42"),
|
|
("a.mov", "video/quicktime", b"\x00\x00\x00\x14ftypqt \x00\x00\x00\x00qt "),
|
|
("a.webm", "video/webm", b"\x1a\x45\xdf\xa3doctypewebm"),
|
|
])
|
|
def test_supported_media_signatures_reach_mail_attachment(service, name, mime, data):
|
|
client, sender, _ = service
|
|
response = client.post("/v1/feedback", files=[
|
|
("submission_id", (None, ID)), ("type", (None, "problem")), ("message", (None, "Issue")),
|
|
("attachments", (name, data, mime)),
|
|
])
|
|
assert response.status_code == 200
|
|
part = list(sender.messages[0].iter_attachments())[0]
|
|
assert part.get_payload(decode=True) == data
|
|
assert part.get_content_type() == mime
|
|
|
|
|
|
def test_expired_metadata_is_removed_at_startup_without_a_new_submission(service):
|
|
client, sender, settings = service
|
|
assert post(client).status_code == 200
|
|
with sqlite3.connect(settings.database_path) as db:
|
|
db.execute("UPDATE submissions SET updated=?", (time.time() - 8 * 86400,))
|
|
create_app(settings, sender)
|
|
with sqlite3.connect(settings.database_path) as db:
|
|
assert db.execute("SELECT COUNT(*) FROM submissions").fetchone()[0] == 0
|
|
|
|
|
|
def test_slow_chunked_upload_times_out_before_sending(service):
|
|
_, sender, settings = service
|
|
settings.upload_timeout_seconds = 0.01
|
|
app = create_app(settings, sender)
|
|
output = []
|
|
|
|
async def receive():
|
|
await asyncio.sleep(0.1)
|
|
return {"type": "http.request", "body": b"", "more_body": True}
|
|
|
|
async def send(event):
|
|
output.append(event)
|
|
|
|
scope = {"type": "http", "asgi": {"version": "3.0"}, "method": "POST",
|
|
"path": "/v1/feedback", "raw_path": b"/v1/feedback", "query_string": b"",
|
|
"headers": [(b"content-type", b"multipart/form-data; boundary=abc")],
|
|
"client": ("127.0.0.1", 123), "server": ("localhost", 80), "scheme": "http"}
|
|
asyncio.run(app(scope, receive, send))
|
|
assert output[0]["status"] == 408
|
|
assert not sender.messages
|
|
|
|
|
|
def test_attachment_filename_extension_matches_detected_type(service):
|
|
client, sender, _ = service
|
|
response = client.post("/v1/feedback", files=[
|
|
("submission_id", (None, ID)), ("type", (None, "problem")), ("message", (None, "Issue")),
|
|
("attachments", ("misleading.mov", PNG, "image/png")),
|
|
])
|
|
assert response.status_code == 200
|
|
assert list(sender.messages[0].iter_attachments())[0].get_filename() == "misleading.png"
|
|
|
|
|
|
def test_absolute_upload_deadline_applies_even_when_chunks_are_ready(service, monkeypatch):
|
|
_, sender, settings = service
|
|
settings.upload_timeout_seconds = 1
|
|
app = create_app(settings, sender)
|
|
clock = iter([0, 0, 0, 2])
|
|
# Replace only this module's clock so the asyncio loop keeps real monotonic time.
|
|
monkeypatch.setattr(feedback_service, "time", SimpleNamespace(
|
|
monotonic=lambda: next(clock, 2), time=time.time))
|
|
events = iter([
|
|
{"type": "http.request", "body": b"first chunk", "more_body": True},
|
|
{"type": "http.request", "body": b"second chunk", "more_body": False},
|
|
])
|
|
output = []
|
|
|
|
async def receive():
|
|
return next(events)
|
|
|
|
async def send(event):
|
|
output.append(event)
|
|
|
|
scope = {"type": "http", "asgi": {"version": "3.0"}, "method": "POST",
|
|
"path": "/v1/feedback", "raw_path": b"/v1/feedback", "query_string": b"",
|
|
"headers": [(b"content-type", b"multipart/form-data; boundary=abc")],
|
|
"client": ("127.0.0.1", 123), "server": ("localhost", 80), "scheme": "http"}
|
|
asyncio.run(app(scope, receive, send))
|
|
assert output[0]["status"] == 408
|
|
assert not sender.messages
|