533 lines
26 KiB
Python
533 lines
26 KiB
Python
"""Feedback submission service. Never log user text, uploads, or SMTP secrets."""
|
||
|
||
import asyncio
|
||
import hashlib
|
||
import hmac
|
||
import json
|
||
import os
|
||
import re
|
||
import smtplib
|
||
import sqlite3
|
||
import ssl
|
||
import tempfile
|
||
import threading
|
||
import time
|
||
from collections import deque
|
||
from contextlib import asynccontextmanager, contextmanager, suppress
|
||
from dataclasses import dataclass, field
|
||
from email.message import EmailMessage
|
||
from email.policy import SMTP
|
||
from email.utils import formatdate
|
||
from pathlib import Path
|
||
|
||
from fastapi import FastAPI, Request
|
||
from starlette.datastructures import UploadFile
|
||
from starlette.exceptions import HTTPException
|
||
from starlette.responses import JSONResponse
|
||
from starlette.concurrency import run_in_threadpool
|
||
|
||
MIB = 1024 * 1024
|
||
FILE_LIMIT = 8 * MIB
|
||
TOTAL_FILE_LIMIT = 20 * MIB
|
||
LOG_LIMIT = 256 * 1024
|
||
BODY_LIMIT = TOTAL_FILE_LIMIT + LOG_LIMIT + 64 * 1024
|
||
ID_PATTERN = re.compile(r"[A-Za-z0-9_-]{16,128}\Z")
|
||
ADDRESS_PATTERN = re.compile(r"[A-Za-z0-9._%+-]+@(?:[A-Za-z0-9-]+\.)+[A-Za-z]{2,63}\Z")
|
||
|
||
|
||
@dataclass
|
||
class Settings:
|
||
database_path: str = "data/feedback.sqlite3"
|
||
smtp_host: str = "smtp.feishu.cn"
|
||
smtp_port: int = 465
|
||
smtp_tls: str = "ssl"
|
||
smtp_username: str = ""
|
||
smtp_password: str = field(default="", repr=False)
|
||
mail_from: str = ""
|
||
mail_to: str = ""
|
||
idempotency_secret: str = field(default="", repr=False)
|
||
smtp_timeout_seconds: int = 15
|
||
retention_days: int = 7
|
||
inflight_stale_seconds: int = 300
|
||
ip_requests_per_hour: int = 10
|
||
global_requests_per_hour: int = 100
|
||
max_concurrent_requests: int = 4
|
||
upload_timeout_seconds: int = 60
|
||
|
||
@classmethod
|
||
def from_env(cls):
|
||
names = {
|
||
"database_path": "DATABASE_PATH", "smtp_host": "SMTP_HOST", "smtp_port": "SMTP_PORT",
|
||
"smtp_tls": "SMTP_TLS", "smtp_username": "SMTP_USERNAME", "smtp_password": "SMTP_PASSWORD",
|
||
"mail_from": "MAIL_FROM", "mail_to": "MAIL_TO", "idempotency_secret": "IDEMPOTENCY_SECRET",
|
||
"smtp_timeout_seconds": "SMTP_TIMEOUT_SECONDS", "retention_days": "RETENTION_DAYS",
|
||
"ip_requests_per_hour": "IP_REQUESTS_PER_HOUR", "global_requests_per_hour": "GLOBAL_REQUESTS_PER_HOUR",
|
||
"max_concurrent_requests": "MAX_CONCURRENT_REQUESTS", "upload_timeout_seconds": "UPLOAD_TIMEOUT_SECONDS",
|
||
}
|
||
defaults, values = cls(), {}
|
||
for name, env_name in names.items():
|
||
if env_name in os.environ:
|
||
value = os.environ[env_name]
|
||
if isinstance(getattr(defaults, name), int):
|
||
try:
|
||
value = int(value)
|
||
except ValueError:
|
||
raise ValueError(f"{env_name} must be an integer") from None
|
||
values[name] = value
|
||
return cls(**values)
|
||
|
||
def validate(self):
|
||
for name in ("mail_from", "mail_to"):
|
||
if not ADDRESS_PATTERN.fullmatch(getattr(self, name)):
|
||
raise ValueError(f"{name} must be one plain email address")
|
||
if not self.smtp_username.strip() or "\r" in self.smtp_username or "\n" in self.smtp_username:
|
||
raise ValueError("SMTP_USERNAME is required and must not contain newlines")
|
||
if not self.smtp_password or self.smtp_password.startswith("REPLACE_"):
|
||
raise ValueError("SMTP_PASSWORD is required")
|
||
if not re.fullmatch(r"[A-Za-z0-9.-]{1,253}", self.smtp_host):
|
||
raise ValueError("SMTP_HOST must be a hostname")
|
||
if (self.smtp_port, self.smtp_tls) not in {(465, "ssl"), (587, "starttls")}:
|
||
raise ValueError("Use SMTP_PORT=465/SMTP_TLS=ssl or SMTP_PORT=587/SMTP_TLS=starttls")
|
||
if len(self.idempotency_secret.encode()) < 32 or self.idempotency_secret.startswith("REPLACE_"):
|
||
raise ValueError("IDEMPOTENCY_SECRET must have at least 32 bytes")
|
||
for name in ("retention_days", "smtp_timeout_seconds", "inflight_stale_seconds",
|
||
"ip_requests_per_hour", "global_requests_per_hour", "max_concurrent_requests",
|
||
"upload_timeout_seconds"):
|
||
if getattr(self, name) <= 0:
|
||
raise ValueError(f"{name} must be positive")
|
||
if self.smtp_timeout_seconds > 30 or self.max_concurrent_requests > 16:
|
||
raise ValueError("SMTP timeout must be <=30 seconds and concurrency <=16")
|
||
|
||
|
||
class ApiError(Exception):
|
||
def __init__(self, http_status, error, detail, *, submission_id=None, status=None):
|
||
self.http_status, self.error, self.detail = http_status, error, detail
|
||
self.submission_id, self.status = submission_id, status
|
||
|
||
|
||
class SendFailure(Exception):
|
||
"""SMTP definitely did not accept this message. Retrying is safe."""
|
||
|
||
|
||
class UncertainFailure(Exception):
|
||
"""DATA may have been accepted, but its acknowledgement was lost."""
|
||
|
||
|
||
def error_response(code, error, detail, *, submission_id=None, status=None, headers=None):
|
||
data = {"success": False, "error": error, "detail": detail}
|
||
if submission_id is not None:
|
||
data["submission_id"] = submission_id
|
||
if status is not None:
|
||
data["status"] = status
|
||
return JSONResponse(data, status_code=code, headers=headers)
|
||
|
||
|
||
class SmtpSender:
|
||
def __init__(self, settings):
|
||
self.settings = settings
|
||
|
||
def __call__(self, message):
|
||
config, client, in_data, accepted = self.settings, None, False, False
|
||
raw_message = message.as_bytes(policy=SMTP)
|
||
try:
|
||
context = ssl.create_default_context()
|
||
if config.smtp_tls == "ssl":
|
||
client = smtplib.SMTP_SSL(config.smtp_host, config.smtp_port,
|
||
timeout=config.smtp_timeout_seconds, context=context)
|
||
else:
|
||
client = smtplib.SMTP(config.smtp_host, config.smtp_port, timeout=config.smtp_timeout_seconds)
|
||
client.ehlo()
|
||
client.starttls(context=context)
|
||
client.ehlo()
|
||
client.login(config.smtp_username, config.smtp_password)
|
||
code, _ = client.mail(config.mail_from)
|
||
if code != 250:
|
||
raise SendFailure()
|
||
code, _ = client.rcpt(config.mail_to)
|
||
if code not in (250, 251):
|
||
raise SendFailure()
|
||
# A lost response from DATA is ambiguous: never automatically resend.
|
||
in_data = True
|
||
code, _ = client.data(raw_message)
|
||
if code != 250:
|
||
raise SendFailure()
|
||
accepted = True
|
||
except SendFailure:
|
||
raise
|
||
except smtplib.SMTPDataError:
|
||
raise SendFailure() from None
|
||
except (OSError, smtplib.SMTPException):
|
||
if in_data:
|
||
raise UncertainFailure() from None
|
||
raise SendFailure() from None
|
||
finally:
|
||
if client is not None:
|
||
try:
|
||
if accepted:
|
||
client.quit()
|
||
else:
|
||
client.close()
|
||
except (OSError, smtplib.SMTPException):
|
||
# QUIT failures do not undo an acknowledged DATA acceptance.
|
||
client.close()
|
||
|
||
|
||
class SubmissionStore:
|
||
def __init__(self, settings):
|
||
self.settings = settings
|
||
Path(settings.database_path).parent.mkdir(parents=True, exist_ok=True)
|
||
with self.connect() as db:
|
||
db.execute("PRAGMA journal_mode=WAL")
|
||
db.execute("CREATE TABLE IF NOT EXISTS submissions ("
|
||
"submission_id TEXT PRIMARY KEY, digest TEXT NOT NULL, status TEXT NOT NULL, "
|
||
"created REAL NOT NULL, updated REAL NOT NULL)")
|
||
db.execute("CREATE INDEX IF NOT EXISTS submissions_updated ON submissions(updated)")
|
||
self.prune()
|
||
|
||
@contextmanager
|
||
def connect(self):
|
||
db = sqlite3.connect(self.settings.database_path, timeout=5)
|
||
try:
|
||
db.execute("PRAGMA synchronous=FULL")
|
||
db.execute("PRAGMA secure_delete=ON")
|
||
with db:
|
||
yield db
|
||
finally:
|
||
db.close()
|
||
|
||
def claim(self, submission_id, digest):
|
||
now = time.time()
|
||
with self.connect() as db:
|
||
db.execute("BEGIN IMMEDIATE")
|
||
db.execute("DELETE FROM submissions WHERE updated < ?",
|
||
(now - self.settings.retention_days * 86400,))
|
||
row = db.execute("SELECT digest, status, updated FROM submissions WHERE submission_id=?",
|
||
(submission_id,)).fetchone()
|
||
if row:
|
||
old_digest, status, updated = row
|
||
if not hmac.compare_digest(old_digest, digest):
|
||
return "conflict"
|
||
if status == "sending":
|
||
if now - updated <= self.settings.inflight_stale_seconds:
|
||
return "in_progress"
|
||
db.execute("UPDATE submissions SET status='uncertain', updated=? WHERE submission_id=?",
|
||
(now, submission_id))
|
||
return "uncertain"
|
||
if status in ("sent", "uncertain"):
|
||
return status
|
||
db.execute("UPDATE submissions SET status='sending', updated=? WHERE submission_id=?",
|
||
(now, submission_id))
|
||
else:
|
||
db.execute("INSERT INTO submissions VALUES (?, ?, 'sending', ?, ?)",
|
||
(submission_id, digest, now, now))
|
||
return "claimed"
|
||
|
||
def finish(self, submission_id, status):
|
||
with self.connect() as db:
|
||
db.execute("UPDATE submissions SET status=?, updated=? WHERE submission_id=?",
|
||
(status, time.time(), submission_id))
|
||
|
||
def prune(self):
|
||
with self.connect() as db:
|
||
db.execute("DELETE FROM submissions WHERE updated < ?",
|
||
(time.time() - self.settings.retention_days * 86400,))
|
||
|
||
|
||
class AdmissionControl:
|
||
"""Bounded per-process admission state: run exactly one worker."""
|
||
def __init__(self, settings):
|
||
self.settings, self.lock, self.current = settings, threading.Lock(), 0
|
||
self.global_times, self.ip_times = deque(), {}
|
||
|
||
def enter(self, ip):
|
||
now = time.monotonic()
|
||
ip_key = hmac.new(self.settings.idempotency_secret.encode(), ip.encode(), hashlib.sha256).digest()
|
||
with self.lock:
|
||
while self.global_times and self.global_times[0] <= now - 3600:
|
||
self.global_times.popleft()
|
||
self.ip_times = {k: times for k, times in self.ip_times.items() if times[-1] > now - 3600}
|
||
times = self.ip_times.get(ip_key, deque())
|
||
while times and times[0] <= now - 3600:
|
||
times.popleft()
|
||
if (len(times) >= self.settings.ip_requests_per_hour or
|
||
len(self.global_times) >= self.settings.global_requests_per_hour):
|
||
return error_response(429, "rate_limited", "Submission limit reached; try again later.",
|
||
headers={"Retry-After": "3600"})
|
||
if self.current >= self.settings.max_concurrent_requests:
|
||
return error_response(503, "server_busy", "Server is busy; retry with the same submission ID.",
|
||
headers={"Retry-After": "15"})
|
||
self.ip_times[ip_key] = times
|
||
times.append(now)
|
||
self.global_times.append(now)
|
||
self.current += 1
|
||
return None
|
||
|
||
def leave(self):
|
||
with self.lock:
|
||
self.current -= 1
|
||
|
||
|
||
class BoundedFeedbackBody:
|
||
"""Count raw bytes before parsing, including chunked bodies and multipart overhead."""
|
||
def __init__(self, app, settings, admission):
|
||
self.app, self.settings, self.admission = app, settings, admission
|
||
|
||
async def __call__(self, scope, receive, send):
|
||
if scope["type"] != "http" or scope.get("path") != "/v1/feedback" or scope["method"] != "POST":
|
||
return await self.app(scope, receive, send)
|
||
denial = self.admission.enter((scope.get("client") or ("unknown",))[0])
|
||
if denial:
|
||
return await denial(scope, receive, send)
|
||
try:
|
||
headers = dict(scope.get("headers", []))
|
||
if headers.get(b"content-type", b"").split(b";", 1)[0].strip().lower() != b"multipart/form-data":
|
||
return await error_response(415, "unsupported_content_type", "Use multipart/form-data.")(scope, receive, send)
|
||
if b"content-length" in headers:
|
||
try:
|
||
length = int(headers[b"content-length"])
|
||
except ValueError:
|
||
return await error_response(400, "invalid_body", "Invalid Content-Length.")(scope, receive, send)
|
||
if length < 0:
|
||
return await error_response(400, "invalid_body", "Invalid Content-Length.")(scope, receive, send)
|
||
if length > BODY_LIMIT:
|
||
return await error_response(413, "upload_too_large", "Request exceeds upload limits.")(scope, receive, send)
|
||
with tempfile.SpooledTemporaryFile(max_size=MIB, mode="w+b") as spool:
|
||
total, deadline = 0, time.monotonic() + self.settings.upload_timeout_seconds
|
||
while True:
|
||
remaining = deadline - time.monotonic()
|
||
if remaining <= 0:
|
||
return await error_response(408, "upload_timeout", "Upload timed out.")(scope, receive, send)
|
||
try:
|
||
event = await asyncio.wait_for(receive(), remaining)
|
||
except TimeoutError:
|
||
return await error_response(408, "upload_timeout", "Upload timed out.")(scope, receive, send)
|
||
if event["type"] == "http.disconnect":
|
||
return
|
||
body = event.get("body", b"")
|
||
total += len(body)
|
||
if total > BODY_LIMIT:
|
||
return await error_response(413, "upload_too_large", "Request exceeds upload limits.")(scope, receive, send)
|
||
spool.write(body)
|
||
if not event.get("more_body", False):
|
||
break
|
||
spool.seek(0)
|
||
replayed = False
|
||
|
||
async def replay_receive():
|
||
nonlocal replayed
|
||
if replayed:
|
||
return {"type": "http.disconnect"}
|
||
block = spool.read(64 * 1024)
|
||
replayed = spool.tell() >= total
|
||
return {"type": "http.request", "body": block, "more_body": not replayed}
|
||
|
||
return await self.app(scope, replay_receive, send)
|
||
finally:
|
||
self.admission.leave()
|
||
|
||
|
||
def file_signature(data, mime):
|
||
if mime == "image/jpeg":
|
||
return data.startswith(b"\xff\xd8\xff"), ".jpg"
|
||
if mime == "image/png":
|
||
return data.startswith(b"\x89PNG\r\n\x1a\n"), ".png"
|
||
if mime == "image/gif":
|
||
return data.startswith((b"GIF87a", b"GIF89a")), ".gif"
|
||
if mime == "image/webp":
|
||
return len(data) >= 12 and data[:4] == b"RIFF" and data[8:12] == b"WEBP", ".webp"
|
||
if mime == "video/webm":
|
||
return data.startswith(b"\x1a\x45\xdf\xa3") and b"webm" in data[:4096], ".webm"
|
||
if mime in {"image/heic", "image/heif", "video/mp4", "video/quicktime", "video/x-m4v"}:
|
||
if len(data) < 16 or data[4:8] != b"ftyp":
|
||
return False, ""
|
||
box_size = int.from_bytes(data[:4], "big")
|
||
if box_size < 16 or box_size > len(data) or box_size > 4096 or box_size % 4:
|
||
return False, ""
|
||
brands = {data[8:12]} | {data[i:i + 4] for i in range(16, box_size, 4)}
|
||
if mime in {"image/heic", "image/heif"}:
|
||
return bool(brands & {b"heic", b"heix", b"hevc", b"hevx", b"heim", b"heis", b"hevm", b"hevs"}), ".heic"
|
||
if mime == "video/quicktime":
|
||
return b"qt " in brands, ".mov"
|
||
return bool(brands & {b"isom", b"iso2", b"avc1", b"mp41", b"mp42", b"M4V ", b"M4VH", b"M4VP", b"dash", b"MSNV"}), ".mp4"
|
||
return False, ""
|
||
|
||
|
||
def safe_filename(filename, extension):
|
||
filename = (filename or "attachment").replace("\\", "/").rsplit("/", 1)[-1]
|
||
filename = re.sub(r"[^A-Za-z0-9._-]", "_", filename).strip("._")[:100] or "attachment"
|
||
alternatives = {".jpg": {".jpg", ".jpeg"}, ".heic": {".heic", ".heif"}, ".mp4": {".mp4", ".m4v"}}
|
||
if Path(filename).suffix.lower() not in alternatives.get(extension, {extension}):
|
||
filename = Path(filename).stem + extension
|
||
return filename
|
||
|
||
|
||
async def parse_submission(request):
|
||
try:
|
||
async with request.form(max_files=4, max_fields=5, max_part_size=20000) as form:
|
||
fields, uploads, diagnostic_upload = {}, [], None
|
||
for name, value in form.multi_items():
|
||
if isinstance(value, UploadFile):
|
||
if name == "diagnostics":
|
||
if diagnostic_upload is not None:
|
||
raise ApiError(422, "validation_error", "Only one diagnostic log is allowed.")
|
||
diagnostic_upload = value
|
||
continue
|
||
if name != "attachments":
|
||
raise ApiError(422, "validation_error", "Unexpected file field.")
|
||
uploads.append(value)
|
||
else:
|
||
if name not in {"submission_id", "type", "message", "contact", "school"} or name in fields:
|
||
raise ApiError(422, "validation_error", "Unexpected or duplicated form field.")
|
||
fields[name] = value
|
||
submission_id = fields.get("submission_id", "")
|
||
if not ID_PATTERN.fullmatch(submission_id):
|
||
raise ApiError(422, "validation_error", "submission_id must be a random 16–128 character identifier.")
|
||
kind = fields.get("type")
|
||
if kind not in {"problem", "development"}:
|
||
raise ApiError(422, "validation_error", "type must be problem or development.")
|
||
for name, maximum in (("message", 5000), ("contact", 200), ("school", 200)):
|
||
if len(fields.get(name, "")) > maximum:
|
||
raise ApiError(422, "validation_error", f"{name} exceeds {maximum} characters.")
|
||
if any("\x00" in text for text in fields.values()):
|
||
raise ApiError(422, "validation_error", "Null characters are not allowed.")
|
||
if kind == "problem":
|
||
if not fields.get("message", "").strip() or "school" in fields:
|
||
raise ApiError(422, "validation_error", "Problem feedback requires message and must omit school.")
|
||
elif ("message" in fields or uploads or diagnostic_upload or not fields.get("school", "").strip()
|
||
or not fields.get("contact", "").strip()):
|
||
raise ApiError(422, "validation_error", "Development requests require school and contact; omit message and attachments.")
|
||
attachments, total = [], 0
|
||
if len(uploads) > 3:
|
||
raise ApiError(413, "upload_too_large", "Maximum three media files.")
|
||
for upload in uploads:
|
||
data = await upload.read(FILE_LIMIT + 1)
|
||
total += len(data)
|
||
if len(data) > FILE_LIMIT or total > TOTAL_FILE_LIMIT:
|
||
raise ApiError(413, "upload_too_large", "Maximum three files, 8 MiB each and 20 MiB total.")
|
||
mime = (upload.content_type or "").lower()
|
||
valid, extension = file_signature(data, mime)
|
||
if not valid:
|
||
raise ApiError(415, "unsupported_media", "Attachment MIME type and file signature must match a supported image or video.")
|
||
attachments.append((safe_filename(upload.filename, extension), mime, data))
|
||
if diagnostic_upload is not None:
|
||
data = await diagnostic_upload.read(LOG_LIMIT + 1)
|
||
if len(data) > LOG_LIMIT:
|
||
raise ApiError(413, "upload_too_large", "Diagnostic log exceeds 256 KiB.")
|
||
try:
|
||
text = data.decode("utf-8")
|
||
except UnicodeDecodeError:
|
||
raise ApiError(415, "unsupported_media", "Diagnostic log must be UTF-8 text.") from None
|
||
if (diagnostic_upload.content_type or '').split(';')[0].strip().lower() != 'text/plain' or '\x00' in text or not text.startswith('TPSchedule diagnostic log\n'):
|
||
raise ApiError(415, "unsupported_media", "Invalid diagnostic log.")
|
||
attachments.append(('TPSchedule-diagnostics.txt', 'text/plain', data))
|
||
return fields, attachments
|
||
except HTTPException as exc:
|
||
detail = str(exc.detail).lower()
|
||
if "too many" in detail or "maximum" in detail or "exceeded maximum size" in detail:
|
||
raise ApiError(413, "upload_too_large", "Multipart field or upload limits exceeded.") from None
|
||
raise ApiError(400, "invalid_body", "Malformed multipart body.") from None
|
||
|
||
|
||
def make_message(settings, fields, attachments):
|
||
message = EmailMessage(policy=SMTP)
|
||
message["From"], message["To"] = settings.mail_from, settings.mail_to
|
||
message["Subject"] = f"[App feedback] {fields['type']} / {fields['submission_id']}"
|
||
message["Message-ID"] = f"<{fields['submission_id']}@{settings.mail_from.split('@')[1]}>"
|
||
message["Date"] = formatdate(localtime=False, usegmt=True)
|
||
lines = [f"Submission ID: {fields['submission_id']}", f"Type: {fields['type']}"]
|
||
if fields["type"] == "development":
|
||
lines += ["School:", fields["school"], "Contact:", fields["contact"]]
|
||
else:
|
||
lines += ["Contact:", fields.get("contact", ""), "Message:", fields["message"]]
|
||
message.set_content("\n".join(lines), charset="utf-8")
|
||
for name, mime, data in attachments:
|
||
main, sub = mime.split("/", 1)
|
||
message.add_attachment(data, maintype=main, subtype=sub, filename=name)
|
||
return message
|
||
|
||
|
||
def create_app(settings=None, sender=None):
|
||
settings = settings or Settings.from_env()
|
||
settings.validate()
|
||
store = SubmissionStore(settings)
|
||
|
||
@asynccontextmanager
|
||
async def lifespan(app):
|
||
async def cleanup():
|
||
while True:
|
||
await asyncio.sleep(3600)
|
||
# Persistent storage failure makes readiness fail instead of exposing data in logs.
|
||
try:
|
||
await run_in_threadpool(store.prune)
|
||
app.state.retention_healthy = True
|
||
except sqlite3.Error:
|
||
app.state.retention_healthy = False
|
||
|
||
app.state.retention_healthy = True
|
||
task = asyncio.create_task(cleanup())
|
||
try:
|
||
yield
|
||
finally:
|
||
task.cancel()
|
||
with suppress(asyncio.CancelledError):
|
||
await task
|
||
|
||
app = FastAPI(docs_url=None, redoc_url=None, openapi_url=None, lifespan=lifespan)
|
||
app.state.sender = sender or SmtpSender(settings)
|
||
app.add_middleware(BoundedFeedbackBody, settings=settings, admission=AdmissionControl(settings))
|
||
|
||
@app.exception_handler(ApiError)
|
||
async def api_error_handler(request, exc):
|
||
return error_response(exc.http_status, exc.error, exc.detail,
|
||
submission_id=exc.submission_id, status=exc.status)
|
||
|
||
@app.get("/healthz")
|
||
async def health():
|
||
# Liveness only: never sends email or claims SMTP readiness.
|
||
if not getattr(app.state, "retention_healthy", True):
|
||
return error_response(503, "storage_unavailable", "Retention cleanup requires operator attention.")
|
||
return {"status": "ok"}
|
||
|
||
@app.post("/v1/feedback")
|
||
async def feedback(request: Request):
|
||
fields, attachments = await parse_submission(request)
|
||
submission_id = fields["submission_id"]
|
||
canonical = json.dumps({"fields": fields, "attachments": [
|
||
{"name": name, "mime": mime, "sha256": hashlib.sha256(data).hexdigest()}
|
||
for name, mime, data in attachments]}, sort_keys=True, ensure_ascii=False, separators=(",", ":"))
|
||
digest = hmac.new(settings.idempotency_secret.encode(), canonical.encode(), hashlib.sha256).hexdigest()
|
||
message = make_message(settings, fields, attachments)
|
||
|
||
def deliver():
|
||
try:
|
||
outcome = store.claim(submission_id, digest)
|
||
if outcome == "conflict":
|
||
raise ApiError(409, "id_conflict", "This ID was already used with different content.", submission_id=submission_id)
|
||
if outcome == "in_progress":
|
||
raise ApiError(409, "in_progress", "A submission with this ID is still being processed.",
|
||
submission_id=submission_id, status="in_progress")
|
||
if outcome == "uncertain":
|
||
raise ApiError(503, "delivery_uncertain", "Mail acceptance is uncertain; do not submit a new ID. Contact the operator.",
|
||
submission_id=submission_id, status="uncertain")
|
||
if outcome == "sent":
|
||
return
|
||
try:
|
||
app.state.sender(message)
|
||
except SendFailure:
|
||
store.finish(submission_id, "retryable")
|
||
raise ApiError(503, "smtp_unavailable", "Mail was not accepted; retry with the same ID.",
|
||
submission_id=submission_id, status="retryable") from None
|
||
except Exception:
|
||
store.finish(submission_id, "uncertain")
|
||
raise ApiError(503, "delivery_uncertain", "Mail acceptance is uncertain; do not submit a new ID. Contact the operator.",
|
||
submission_id=submission_id, status="uncertain") from None
|
||
store.finish(submission_id, "sent")
|
||
except sqlite3.Error:
|
||
raise ApiError(503, "storage_unavailable", "Submission state could not be confirmed; retry only with the same ID.",
|
||
submission_id=submission_id, status="uncertain") from None
|
||
|
||
await run_in_threadpool(deliver)
|
||
return {"success": True, "submission_id": submission_id, "status": "sent"}
|
||
|
||
return app
|