From b741389754604bcfd3675779687cf3b578a5390c Mon Sep 17 00:00:00 2001 From: Philipp Wagner Date: Thu, 29 Oct 2026 14:08:16 +0100 Subject: [PATCH] fix: Debug-Modus raus, gunicorn, UIDs statt Sequenznummern, nur http(s)-Links MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Flask-Debugger (Werkzeug, Codeausführung) war offen auf 0.0.0.0:5000; jetzt gunicorn, Healthcheck - Verschieben per UID (Sequenznummern verschieben sich nach dem Expunge), nur Ziffern erlaubt - "Move All" schickte keine IDs mit; Formular enthält jetzt alle UIDs - Links aus Mails nur noch http(s) (kein javascript:/data:), rel=noopener noreferrer - Betreff dekodiert, BODY.PEEK (Mails bleiben ungelesen), Python 3.12, feste Versionen, Nicht-Root - Tests Claude-Session: https://claude.ai/code/session_01SqzF6qT3TQ8ej3wigEuMSu --- Dockerfile | 19 +++-- README.md | 8 ++ app.py | 211 +++++++++++++++++++++++++--------------------- requirements.txt | 7 +- result.html | 7 +- tests/test_app.py | 86 +++++++++++++++++++ 6 files changed, 227 insertions(+), 111 deletions(-) create mode 100644 tests/test_app.py diff --git a/Dockerfile b/Dockerfile index c6d5c7a..66d1f8c 100644 --- a/Dockerfile +++ b/Dockerfile @@ -1,18 +1,19 @@ -# Verwende das offizielle Python-Image als Basis -FROM python:3.8-slim +FROM python:3.12-slim -# Setze das Arbeitsverzeichnis innerhalb des Containers WORKDIR /app -# Installiere die erforderlichen Abhängigkeiten COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt -# Kopiere die Anwendungsdateien in den Container -COPY . . +COPY app.py index.html result.html ./ +COPY static ./static -# Starte die Anwendung, wenn der Container gestartet wird -CMD ["python", "app.py"] +RUN useradd -m -u 1000 app +USER app -#Expose den Port 5000 (Nur Dokumentation für Entwickler) EXPOSE 5000 +HEALTHCHECK --interval=30s --timeout=5s --start-period=10s \ + CMD python -c "import urllib.request; urllib.request.urlopen('http://127.0.0.1:5000/health')" + +# gunicorn instead of the Flask development server (no Werkzeug debugger) +CMD ["gunicorn", "--bind", "0.0.0.0:5000", "--workers", "2", "--timeout", "120", "app:app"] diff --git a/README.md b/README.md index e5c9285..ccc19d4 100644 --- a/README.md +++ b/README.md @@ -22,3 +22,11 @@ Run following command: ```docker run -p 5000:5000 -e EMAIL_USERNAME=your_email@e Copyright 2023 Philipp Wagner This project is licensed under the GPLv3 License - see the [LICENSE](LICENSE) file for details. + +## Betrieb + +Konfiguration per Umgebung: `EMAIL_SERVER`, `EMAIL_PORT` (993), `EMAIL_USERNAME`, `EMAIL_PASSWORD` +(Sammelpostfach), optional `EMAIL_FOLDER` (`INBOX`) und `TRASH_FOLDER` (`Trash`). +Der Container startet gunicorn auf Port 5000 (kein Flask-Debugger) und gehört hinter einen +Reverse Proxy mit Anmeldung (bei Wagnersnetz Caddy `forward_auth` über Authelia). +Tests: `pytest tests`. diff --git a/app.py b/app.py index 1660b3c..b6bfc6e 100644 --- a/app.py +++ b/app.py @@ -1,138 +1,155 @@ -import imaplib +""" +E-Mail Unsubscribe Link Finder. + +Reads a collection mailbox (forwarded newsletters), lists the unsubscribe links per mail and moves +handled mails to the trash. Runs behind the reverse proxy (Caddy forward_auth / Authelia); the +mailbox credentials come from the environment. +""" + import email -import re +import imaplib +import logging import os -from flask import Flask, render_template, request, redirect, url_for -from dotenv import load_dotenv +import re +from email.header import decode_header, make_header + from bs4 import BeautifulSoup +from dotenv import load_dotenv +from flask import Flask, redirect, render_template, request, url_for -app = Flask(__name__) -load_dotenv() # Laden der Umgebungsvariablen aus der .env-Datei +load_dotenv() # optional .env for local runs -# Email-Server-Konfiguration -EMAIL_USERNAME = os.environ.get("EMAIL_USERNAME") -EMAIL_PASSWORD = os.environ.get("EMAIL_PASSWORD") -EMAIL_SERVER = os.environ.get("EMAIL_SERVER") +EMAIL_USERNAME = os.environ.get("EMAIL_USERNAME", "") +EMAIL_PASSWORD = os.environ.get("EMAIL_PASSWORD", "") +EMAIL_SERVER = os.environ.get("EMAIL_SERVER", "") +EMAIL_PORT = int(os.environ.get("EMAIL_PORT", "993")) +EMAIL_FOLDER = os.environ.get("EMAIL_FOLDER", "INBOX") +TRASH_FOLDER = os.environ.get("TRASH_FOLDER", "Trash") + +logging.basicConfig(level=os.environ.get("LOG_LEVEL", "INFO")) +logger = logging.getLogger(__name__) template_dir = os.path.abspath(os.path.dirname(__file__)) app = Flask(__name__, template_folder=template_dir) -link_pattern = r"(.*?)" -unsubscribe_pattern = r"(?i)\babbestellen\b|\bunsubscribe\b|\bdeabonnieren\b|\babbestellung\b|\babmelden\b" -def get_text_from_email(email_message): +UNSUBSCRIBE_PATTERN = re.compile( + r"\babbestellen\b|\bunsubscribe\b|\bdeabonnieren\b|\babbestellung\b|\babmelden\b", re.IGNORECASE +) +URL_PATTERN = re.compile(r"https?://[^\s<>\"']+") +UID_PATTERN = re.compile(r"^\d+$") + + +def is_safe_link(url: str) -> bool: + """Only plain web links: mails are untrusted, javascript:/data: links must never become clickable.""" + return url.lower().startswith(("https://", "http://")) + + +def get_text_from_email(message) -> str: text = "" - for part in email_message.walk(): - content_type = part.get_content_type() - content_disposition = str(part.get("Content-Disposition")) - - if "text" in content_type and "attachment" not in content_disposition: + for part in message.walk(): + disposition = str(part.get("Content-Disposition")) + if "text" in part.get_content_type() and "attachment" not in disposition: body = part.get_payload(decode=True) if body: text += body.decode(errors="ignore") return text -def find_unsubscribe_links(text, html_links): - soup = BeautifulSoup(text, 'html.parser') - unsubscribe_links = [] + +def get_subject(message) -> str: + raw = message.get("Subject", "") + try: + return str(make_header(decode_header(raw))) + except Exception: + return raw - for link in soup.find_all('a', href=True): - if re.search(unsubscribe_pattern, str(link)) and str(link['href']) not in html_links: - print(link['href']) - test1=str(link['href']) - print(test1) - if test1 not in unsubscribe_links: - print(test1) - unsubscribe_links.append(test1) - +def find_unsubscribe_links(text: str) -> list[str]: + """Unsubscribe links from HTML anchors (link text or URL matches) and from plain-text URLs.""" + links: list[str] = [] + soup = BeautifulSoup(text, "html.parser") + for anchor in soup.find_all("a", href=True): + href = str(anchor["href"]).strip() + if UNSUBSCRIBE_PATTERN.search(str(anchor)) and is_safe_link(href) and href not in links: + links.append(href) + for url in URL_PATTERN.findall(text): + if UNSUBSCRIBE_PATTERN.search(url) and url not in links: + links.append(url) + return links - # Zusätzlich nach einfachen URLs suchen und auf das Unsubscribe-Pattern prüfen - simple_urls = re.findall(r"(https?://\S+)", text) - for url in simple_urls: - if re.search(unsubscribe_pattern, url) and url not in html_links: - if url not in unsubscribe_links: - unsubscribe_links.append(url) - return unsubscribe_links +def connect() -> imaplib.IMAP4_SSL: + mail = imaplib.IMAP4_SSL(EMAIL_SERVER, EMAIL_PORT) + mail.login(EMAIL_USERNAME, EMAIL_PASSWORD) + mail.select(EMAIL_FOLDER) + return mail -def get_subject_from_email(email_message): - subject = email_message.get("Subject") - return subject -def move_emails_to_trash(email_ids): +def collect_links(mail) -> dict: + """{uid: {"subject": ..., "links": [...]}} for all mails with unsubscribe links (UIDs stay stable).""" + _, data = mail.uid("search", None, "ALL") + result = {} + for uid in data[0].split(): + _, msg_data = mail.uid("fetch", uid, "(BODY.PEEK[])") + if not msg_data or not isinstance(msg_data[0], tuple): + continue + message = email.message_from_bytes(msg_data[0][1]) + body = get_text_from_email(message) + links = find_unsubscribe_links(body) if body else [] + if links: + result[uid.decode()] = {"subject": get_subject(message), "links": links} + return result - username = EMAIL_USERNAME - password = EMAIL_PASSWORD - server = EMAIL_SERVER - mail = imaplib.IMAP4_SSL(server) - mail.login(username, password) - mail.select("inbox") +def move_emails_to_trash(uids: list[str]) -> int: + uids = [u for u in uids if UID_PATTERN.match(u)] + if not uids: + return 0 + mail = connect() + try: + for uid in uids: + # Mailcow/Dovecot: copy to trash, flag deleted, expunge + mail.uid("copy", uid, TRASH_FOLDER) + mail.uid("store", uid, "+FLAGS", "(\\Deleted)") + mail.expunge() + finally: + mail.logout() + logger.info("Moved %d mail(s) to %s", len(uids), TRASH_FOLDER) + return len(uids) - for email_id in email_ids: - # Verschiebe die E-Mail in den Papierkorb (Trash) - mail.copy(email_id, "Trash") - # Setze die \Deleted-Flagge, um die E-Mail später zu löschen (für Mailcow) - mail.store(email_id, "+FLAGS", "\\Deleted") - # Speichere die Änderungen auf dem IMAP-Server - mail.expunge() - mail.logout() - -@app.route("/", methods=["GET", "POST"]) +@app.route("/", methods=["GET"]) def index(): return render_template("index.html", email_address=EMAIL_USERNAME) + @app.route("/result") def result(): - - username = EMAIL_USERNAME - password = EMAIL_PASSWORD - server = EMAIL_SERVER - - mail = imaplib.IMAP4_SSL(server) - mail.login(username, password) - mail.select("inbox") - - _, messages = mail.search(None, "ALL") - links_data = {} - html_links = set() - - for message_num in messages[0].split(): - _, msg_data = mail.fetch(message_num, "(RFC822)") - msg = email.message_from_bytes(msg_data[0][1]) - body = get_text_from_email(msg) - subject = get_subject_from_email(msg) - if body: - unsubscribe_links_in_html = find_unsubscribe_links(body, html_links) - html_links.update(unsubscribe_links_in_html) - unsubscribe_links_in_text = find_unsubscribe_links(body, html_links) - all_unsubscribe_links = list(set(unsubscribe_links_in_html + unsubscribe_links_in_text)) - if all_unsubscribe_links: - links_data[message_num.decode()] = { - "subject": subject, - "links": all_unsubscribe_links - } - - mail.logout() - + mail = connect() + try: + links_data = collect_links(mail) + finally: + mail.logout() return render_template("result.html", links_data=links_data) + @app.route("/move-to-trash", methods=["POST"]) def move_to_trash(): - if request.method == "POST": - email_id = request.form.get("email_id") - if email_id: - move_emails_to_trash([email_id]) + email_id = request.form.get("email_id", "") + move_emails_to_trash([email_id]) return redirect(url_for("result")) + @app.route("/move-all-to-trash", methods=["POST"]) def move_all_to_trash(): - if request.method == "POST": - email_ids = request.form.getlist("email_id") - if email_ids: - move_emails_to_trash(email_ids) + move_emails_to_trash(request.form.getlist("email_id")) return redirect(url_for("index")) + +@app.route("/health") +def health(): + return {"status": "ok"} + + if __name__ == "__main__": - app.run(host="0.0.0.0", port=5000, debug=True) \ No newline at end of file + # local development only; the container runs gunicorn (no debugger) + app.run(host="127.0.0.1", port=5000, debug=os.environ.get("FLASK_DEBUG") == "1") diff --git a/requirements.txt b/requirements.txt index 78b9713..68fc90a 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,3 +1,4 @@ -Flask -python-dotenv -beautifulsoup4 \ No newline at end of file +Flask==3.1.2 +python-dotenv==1.2.1 +beautifulsoup4==4.14.2 +gunicorn==23.0.0 diff --git a/result.html b/result.html index 71e7af9..85ae409 100644 --- a/result.html +++ b/result.html @@ -9,7 +9,7 @@

E-Mail Unsubscribe Links Result

- + @@ -20,7 +20,7 @@
Email IDUID Subject Unsubscribe Link Move to Trash{{ data.subject }} {% for link in data.links %} - {{ link[:30] + '...' }}
+ {{ link[:30] + '...' }}
{% endfor %} 		@@ -36,6 +36,9 @@
+ {% for email_id in links_data %} + + {% endfor %}
diff --git a/tests/test_app.py b/tests/test_app.py new file mode 100644 index 0000000..310b087 --- /dev/null +++ b/tests/test_app.py @@ -0,0 +1,86 @@ +"""Tests for link extraction, link safety and the trash actions (fake IMAP server).""" + +from email.message import EmailMessage + +import pytest + +import app as unsub + + +def _mail(html: str, subject: str = "Newsletter") -> bytes: + msg = EmailMessage() + msg["Subject"] = subject + msg["From"] = "news@example.org" + msg.set_content("plain") + msg.add_alternative(html, subtype="html") + return msg.as_bytes() + + +def test_finds_anchor_and_plain_links_without_duplicates(): + html = ( + 'Abmelden Home' + " https://ex.org/unsubscribe/2 https://ex.org/unsubscribe/2" + ) + assert unsub.find_unsubscribe_links(html) == ["https://ex.org/u?id=1", "https://ex.org/unsubscribe/2"] + + +@pytest.mark.parametrize("href", ["javascript:alert(1)", "data:text/html,", "JavaScript:x"]) +def test_dangerous_links_are_dropped(href): + assert unsub.find_unsubscribe_links(f'Unsubscribe') == [] + + +def test_encoded_subject_is_decoded(): + msg = EmailMessage() + msg["Subject"] = "Grüße vom Newsletter" + parsed = unsub.email.message_from_bytes(msg.as_bytes()) + assert unsub.get_subject(parsed) == "Grüße vom Newsletter" + + +class FakeIMAP: + def __init__(self, mails: dict[bytes, bytes]): + self.mails = mails + self.calls = [] + + def uid(self, command, *args): + self.calls.append((command, args)) + if command == "search": + return "OK", [b" ".join(self.mails)] + if command == "fetch": + return "OK", [(b"1 (BODY[] {n}", self.mails[args[0]]), b")"] + return "OK", [b""] + + def expunge(self): + self.calls.append(("expunge", ())) + + def logout(self): + pass + + +def test_collect_links_keys_by_uid(): + fake = FakeIMAP({b"41": _mail('unsubscribe'), b"42": _mail("

no links

")}) + result = unsub.collect_links(fake) + assert list(result) == ["41"] and result["41"]["links"] == ["https://ex.org/u"] + + +def test_move_to_trash_uses_uids_and_ignores_garbage(monkeypatch): + fake = FakeIMAP({}) + monkeypatch.setattr(unsub, "connect", lambda: fake) + assert unsub.move_emails_to_trash(["7", "1:*", "8 OR ALL"]) == 1 + assert ("copy", ("7", "Trash")) in fake.calls + assert ("store", ("7", "+FLAGS", "(\\Deleted)")) in fake.calls + assert not any(c[0] in ("copy", "store") and c[1][0] != "7" for c in fake.calls) + + +def test_move_all_form_posts_every_uid(monkeypatch): + moved = [] + monkeypatch.setattr(unsub, "move_emails_to_trash", lambda uids: moved.extend(uids) or len(uids)) + client = unsub.app.test_client() + assert client.post("/move-all-to-trash", data={"email_id": ["41", "42"]}).status_code == 302 + assert moved == ["41", "42"] + + +def test_result_page_renders_hidden_ids_for_move_all(monkeypatch): + monkeypatch.setattr(unsub, "connect", lambda: FakeIMAP({b"41": _mail('Abmelden')})) + html = unsub.app.test_client().get("/result").get_data(as_text=True) + assert html.count('name="email_id" value="41"') == 2 # single form + move-all form + assert 'rel="noopener noreferrer"' in html