Add feedback service, reminder diagnostics and smooth week transitions
This commit is contained in:
Codex
@@ -0,0 +1,513 @@
|
||||
"""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
|
||||
BODY_LIMIT = TOTAL_FILE_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=3, max_fields=5, max_part_size=20000) as form:
|
||||
fields, uploads = {}, []
|
||||
for name, value in form.multi_items():
|
||||
if isinstance(value, UploadFile):
|
||||
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 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
|
||||
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))
|
||||
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
|
||||
Reference in New Issue
Block a user