From b741389754604bcfd3675779687cf3b578a5390c Mon Sep 17 00:00:00 2001 From: Philipp Wagner Date: Thu, 29 Oct 2026 14:08:16 +0100 Subject: [PATCH 1/2] 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 -- 2.54.0 From 68940e831a8ad115e7333423e3b2f37001a14f1c Mon Sep 17 00:00:00 2001 From: Philipp Wagner Date: Thu, 29 Oct 2026 14:29:22 +0100 Subject: [PATCH 2/2] feat: Sammelpostfach per SSO (OAuth/XOAUTH2) statt Passwort MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Einmal "Connect mailbox" (OIDC mit PKCE, prompt=login, Anmeldung als technisches Konto des Postfachs), Prüfung der Mail-Adresse im Token, Refresh-Token verschlüsselt in DATA_DIR, Access-Token im Speicher; ein gunicorn-Worker wegen der Token-Rotation. Ohne OIDC_* weiter mit EMAIL_PASSWORD. Claude-Session: https://claude.ai/code/session_01SqzF6qT3TQ8ej3wigEuMSu --- Dockerfile | 7 +- README.md | 4 + app.py | 58 +++++++++++++- index.html | 14 ++++ oauth.py | 184 ++++++++++++++++++++++++++++++++++++++++++++ requirements.txt | 2 + tests/test_oauth.py | 141 +++++++++++++++++++++++++++++++++ 7 files changed, 404 insertions(+), 6 deletions(-) create mode 100644 oauth.py create mode 100644 tests/test_oauth.py diff --git a/Dockerfile b/Dockerfile index 66d1f8c..248a33f 100644 --- a/Dockerfile +++ b/Dockerfile @@ -5,7 +5,7 @@ WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt -COPY app.py index.html result.html ./ +COPY app.py oauth.py index.html result.html ./ COPY static ./static RUN useradd -m -u 1000 app @@ -15,5 +15,6 @@ 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"] +# gunicorn instead of the Flask development server (no Werkzeug debugger); one worker, because the +# OAuth refresh token rotates and must not be refreshed by two processes at once +CMD ["gunicorn", "--bind", "0.0.0.0:5000", "--workers", "1", "--threads", "4", "--timeout", "120", "app:app"] diff --git a/README.md b/README.md index ccc19d4..382006e 100644 --- a/README.md +++ b/README.md @@ -27,6 +27,10 @@ This project is licensed under the GPLv3 License - see the [LICENSE](LICENSE) fi Konfiguration per Umgebung: `EMAIL_SERVER`, `EMAIL_PORT` (993), `EMAIL_USERNAME`, `EMAIL_PASSWORD` (Sammelpostfach), optional `EMAIL_FOLDER` (`INBOX`) und `TRASH_FOLDER` (`Trash`). +Statt `EMAIL_PASSWORD` geht SSO (OAuth/XOAUTH2): `OIDC_ISSUER`, `OIDC_CLIENT_ID`, `OIDC_CLIENT_SECRET`, +`OIDC_REDIRECT_URI` (`https:///oidc/callback`), `TOKEN_KEY` (Fernet-Schlüssel) und `DATA_DIR` (`/data`). +Dann zeigt die Startseite „Connect mailbox“: einmal im privaten Fenster als technisches Konto des +Sammelpostfachs am SSO anmelden; das Refresh-Token liegt verschlüsselt in `DATA_DIR/oauth-token`. 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 b6bfc6e..7bac0e1 100644 --- a/app.py +++ b/app.py @@ -15,7 +15,9 @@ 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 +from flask import Flask, abort, redirect, render_template, request, url_for + +from oauth import MailboxOAuth, OAuthError, xoauth2_string load_dotenv() # optional .env for local runs @@ -32,6 +34,9 @@ logger = logging.getLogger(__name__) template_dir = os.path.abspath(os.path.dirname(__file__)) app = Flask(__name__, template_folder=template_dir) +# OAuth (SSO) for the collection mailbox if OIDC_* and TOKEN_KEY are set; otherwise EMAIL_PASSWORD +OAUTH = MailboxOAuth.from_env() + UNSUBSCRIBE_PATTERN = re.compile( r"\babbestellen\b|\bunsubscribe\b|\bdeabonnieren\b|\babbestellung\b|\babmelden\b", re.IGNORECASE ) @@ -78,8 +83,12 @@ def find_unsubscribe_links(text: str) -> list[str]: def connect() -> imaplib.IMAP4_SSL: + token = OAUTH.access_token() if OAUTH else None # before connecting: fail fast without a token mail = imaplib.IMAP4_SSL(EMAIL_SERVER, EMAIL_PORT) - mail.login(EMAIL_USERNAME, EMAIL_PASSWORD) + if token: + mail.authenticate("XOAUTH2", lambda _: xoauth2_string(EMAIL_USERNAME, token).encode()) + else: + mail.login(EMAIL_USERNAME, EMAIL_PASSWORD) mail.select(EMAIL_FOLDER) return mail @@ -119,7 +128,50 @@ def move_emails_to_trash(uids: list[str]) -> int: @app.route("/", methods=["GET"]) def index(): - return render_template("index.html", email_address=EMAIL_USERNAME) + return render_template( + "index.html", + email_address=EMAIL_USERNAME, + oauth=OAUTH is not None, + connected=OAUTH.connected() if OAUTH else True, + error=request.args.get("error", ""), + ) + + +@app.route("/connect") +def connect_mailbox(): + """Connect the collection mailbox: sign in at the SSO as its technical account (private window).""" + if not OAUTH: + abort(404) + return redirect(OAUTH.start()) + + +@app.route("/oidc/callback") +def oidc_callback(): + if not OAUTH: + abort(404) + if request.args.get("error"): + return redirect(url_for("index", error=request.args["error"])) + try: + OAUTH.finish(request.args.get("state", ""), request.args.get("code", "")) + except OAuthError as e: + logger.warning("Mailbox connect failed: %s", e) + return redirect(url_for("index", error=str(e))) + logger.info("Collection mailbox %s connected via OAuth", EMAIL_USERNAME) + return redirect(url_for("index")) + + +@app.route("/disconnect", methods=["POST"]) +def disconnect_mailbox(): + if not OAUTH: + abort(404) + OAUTH.disconnect() + return redirect(url_for("index")) + + +@app.errorhandler(OAuthError) +def oauth_error(e): + logger.warning("Mailbox access failed: %s", e) + return redirect(url_for("index", error="Mailbox not connected or token expired – connect again")) @app.route("/result") diff --git a/index.html b/index.html index a65b39f..d8ba050 100644 --- a/index.html +++ b/index.html @@ -8,9 +8,23 @@

E-Mail Unsubscribe Link Finder

Email your emails to: {{ email_address }}

+ {% if error %}

{{ error }}

{% endif %} + {% if connected %}
+ {% if oauth %} +
+ +
+ {% endif %} + {% else %} +

The collection mailbox is not connected yet. Open this page in a private window and sign in + as {{ email_address }} (technical account) – not with your own account.

+
+ +
+ {% endif %}
diff --git a/oauth.py b/oauth.py new file mode 100644 index 0000000..670a192 --- /dev/null +++ b/oauth.py @@ -0,0 +1,184 @@ +""" +OAuth access to the collection mailbox (XOAUTH2) instead of a stored password. + +An admin connects the mailbox once: OIDC code flow with PKCE, signing in at the SSO as the technical +account of the mailbox. The refresh token is stored encrypted in DATA_DIR; access tokens stay in memory. +""" + +import base64 +import hashlib +import json +import os +import secrets +import threading +import time +from pathlib import Path +from urllib.parse import urlencode + +import httpx +from cryptography.fernet import Fernet, InvalidToken + +SCOPES = "openid profile email offline_access" +REFRESH_MARGIN_SEC = 60 +PENDING_TTL_SEC = 600 + + +class OAuthError(Exception): + pass + + +def _b64url(data: bytes) -> str: + return base64.urlsafe_b64encode(data).rstrip(b"=").decode() + + +def new_pkce() -> tuple[str, str]: + verifier = _b64url(secrets.token_bytes(48)) + return verifier, _b64url(hashlib.sha256(verifier.encode()).digest()) + + +def jwt_claims(token: str) -> dict: + try: + payload = token.split(".")[1] + return json.loads(base64.urlsafe_b64decode(payload + "=" * (-len(payload) % 4))) + except (IndexError, ValueError) as e: + raise OAuthError("Access token is not a JWT") from e + + +def xoauth2_string(user: str, token: str) -> str: + return f"user={user}\x01auth=Bearer {token}\x01\x01" + + +class MailboxOAuth: + """Token handling for exactly one mailbox.""" + + def __init__(self, issuer, client_id, client_secret, redirect_uri, mailbox, data_dir, key, + transport=None, clock=time.time): + self.client_id = client_id + self.client_secret = client_secret + self.redirect_uri = redirect_uri + self.mailbox = mailbox + self.authorization_endpoint = issuer.rstrip("/") + "/api/oidc/authorization" + self.token_endpoint = issuer.rstrip("/") + "/api/oidc/token" + self.token_file = Path(data_dir) / "oauth-token" + self.cipher = Fernet(key.encode() if isinstance(key, str) else key) + self.transport = transport + self.clock = clock + self._lock = threading.Lock() + self._access: tuple[str, float] | None = None + self._pending: dict[str, tuple[str, float]] = {} + + @classmethod + def from_env(cls): + names = ["OIDC_ISSUER", "OIDC_CLIENT_ID", "OIDC_CLIENT_SECRET", "OIDC_REDIRECT_URI", "TOKEN_KEY"] + if not all(os.environ.get(n) for n in names): + return None + return cls( + issuer=os.environ["OIDC_ISSUER"], + client_id=os.environ["OIDC_CLIENT_ID"], + client_secret=os.environ["OIDC_CLIENT_SECRET"], + redirect_uri=os.environ["OIDC_REDIRECT_URI"], + mailbox=os.environ.get("EMAIL_USERNAME", ""), + data_dir=os.environ.get("DATA_DIR", "/data"), + key=os.environ["TOKEN_KEY"], + ) + + # --- stored refresh token --------------------------------------------------------------- + def connected(self) -> bool: + return self._read_refresh() is not None + + def _read_refresh(self) -> str | None: + try: + return self.cipher.decrypt(self.token_file.read_bytes()).decode() + except (FileNotFoundError, InvalidToken): + return None + + def _write_refresh(self, token: str | None) -> None: + if token is None: + self.token_file.unlink(missing_ok=True) + return + tmp = self.token_file.with_suffix(".tmp") + tmp.write_bytes(self.cipher.encrypt(token.encode())) + tmp.chmod(0o600) + tmp.replace(self.token_file) + + # --- connect flow ------------------------------------------------------------------------- + def start(self) -> str: + verifier, challenge = new_pkce() + state = secrets.token_urlsafe(32) + now = self.clock() + with self._lock: + self._pending = {k: v for k, v in self._pending.items() if now - v[1] < PENDING_TTL_SEC} + self._pending[state] = (verifier, now) + query = { + "response_type": "code", + "client_id": self.client_id, + "redirect_uri": self.redirect_uri, + "scope": SCOPES, + "state": state, + "code_challenge": challenge, + "code_challenge_method": "S256", + "prompt": "login", # sign in as the mailbox account, not with the admin's session + "login_hint": self.mailbox, + } + return self.authorization_endpoint + "?" + urlencode(query) + + def finish(self, state: str, code: str) -> None: + with self._lock: + entry = self._pending.pop(state, None) + if not entry or self.clock() - entry[1] > PENDING_TTL_SEC or not code: + raise OAuthError("Invalid or expired request") + tokens = self._token_request({ + "grant_type": "authorization_code", + "code": code, + "redirect_uri": self.redirect_uri, + "code_verifier": entry[0], + }) + email = jwt_claims(tokens["access_token"]).get("email", "") + if email.lower() != self.mailbox.lower(): + raise OAuthError(f"Signed in as {email or 'unknown'}, expected {self.mailbox}") + if not tokens.get("refresh_token"): + raise OAuthError("SSO returned no refresh token") + with self._lock: + self._store(tokens) + + def disconnect(self) -> None: + with self._lock: + self._write_refresh(None) + self._access = None + + # --- tokens ------------------------------------------------------------------------------- + def _token_request(self, data: dict) -> dict: + try: + with httpx.Client(timeout=20.0, transport=self.transport) as client: + r = client.post(self.token_endpoint, data=data, auth=(self.client_id, self.client_secret)) + except httpx.HTTPError as e: + raise OAuthError(f"Token endpoint unreachable: {e}") from e + try: + body = r.json() + except ValueError: + body = {} + if r.status_code != 200 or not body.get("access_token"): + raise OAuthError(f"Token request failed ({r.status_code}): {body.get('error', 'unknown')}") + return body + + def _store(self, tokens: dict) -> None: + if tokens.get("refresh_token"): + self._write_refresh(tokens["refresh_token"]) + self._access = (tokens["access_token"], self.clock() + float(tokens.get("expires_in") or 300)) + + def access_token(self) -> str: + with self._lock: + if self._access and self._access[1] - REFRESH_MARGIN_SEC > self.clock(): + return self._access[0] + refresh = self._read_refresh() + if not refresh: + raise OAuthError("Mailbox not connected") + try: + tokens = self._token_request({"grant_type": "refresh_token", "refresh_token": refresh}) + except OAuthError as e: + if "invalid_grant" in str(e): + self._write_refresh(None) # revoked/expired: connect again + self._access = None + raise + self._store(tokens) + return tokens["access_token"] diff --git a/requirements.txt b/requirements.txt index 68fc90a..8d32d79 100644 --- a/requirements.txt +++ b/requirements.txt @@ -2,3 +2,5 @@ Flask==3.1.2 python-dotenv==1.2.1 beautifulsoup4==4.14.2 gunicorn==23.0.0 +httpx==0.28.1 +cryptography==46.0.3 diff --git a/tests/test_oauth.py b/tests/test_oauth.py new file mode 100644 index 0000000..d9c4298 --- /dev/null +++ b/tests/test_oauth.py @@ -0,0 +1,141 @@ +"""Tests for the OAuth connection of the collection mailbox.""" + +import base64 +import json +from urllib.parse import parse_qs, urlparse + +import httpx +import pytest +from cryptography.fernet import Fernet + +import oauth + + +def _jwt(claims): + enc = lambda d: base64.urlsafe_b64encode(json.dumps(d).encode()).rstrip(b"=").decode() # noqa: E731 + return f"{enc({'alg': 'RS256'})}.{enc(claims)}.sig" + + +class Clock: + now = 1000.0 + + def __call__(self): + return self.now + + +def _mb(tmp_path, responses, seen=None, clock=None): + def handler(request): + if seen is not None: + seen.append(parse_qs(request.content.decode())) + return responses.pop(0) + return oauth.MailboxOAuth( + "https://sso.example.org", "unsubscriber", "secret", "https://u.example.org/oidc/callback", + "unsubscribe-bot@example.org", tmp_path, Fernet.generate_key().decode(), + transport=httpx.MockTransport(handler), clock=clock or Clock(), + ) + + +def _start_state(mb): + url = urlparse(mb.start()) + q = parse_qs(url.query) + assert q["prompt"] == ["login"] and q["login_hint"] == ["unsubscribe-bot@example.org"] + assert "offline_access" in q["scope"][0] and q["code_challenge_method"] == ["S256"] + return q["state"][0] + + +def test_connect_stores_encrypted_refresh_token(tmp_path): + mb = _mb(tmp_path, [httpx.Response(200, json={ + "access_token": _jwt({"email": "Unsubscribe-Bot@example.org"}), "refresh_token": "rt-1", "expires_in": 3600})]) + mb.finish(_start_state(mb), "code") + assert mb.connected() + assert b"rt-1" not in (tmp_path / "oauth-token").read_bytes() + assert oct((tmp_path / "oauth-token").stat().st_mode)[-3:] == "600" + + +def test_connect_with_wrong_account_is_rejected(tmp_path): + mb = _mb(tmp_path, [httpx.Response(200, json={ + "access_token": _jwt({"email": "philipp@example.org"}), "refresh_token": "rt", "expires_in": 3600})]) + with pytest.raises(oauth.OAuthError, match="expected unsubscribe-bot"): + mb.finish(_start_state(mb), "code") + assert not mb.connected() + + +def test_unknown_state_is_rejected(tmp_path): + with pytest.raises(oauth.OAuthError): + _mb(tmp_path, []).finish("forged", "code") + + +def test_access_token_refreshes_and_rotates(tmp_path): + seen, clock = [], Clock() + mb = _mb(tmp_path, [ + httpx.Response(200, json={"access_token": _jwt({"email": "unsubscribe-bot@example.org"}), + "refresh_token": "rt-1", "expires_in": 3600}), + httpx.Response(200, json={"access_token": "at-2", "refresh_token": "rt-2", "expires_in": 3600}), + ], seen, clock) + mb.finish(_start_state(mb), "code") + first = mb.access_token() + assert len(seen) == 1 # cached from the code exchange + clock.now += 3600 + assert mb.access_token() == "at-2" != first + assert seen[1]["refresh_token"] == ["rt-1"] + assert mb._read_refresh() == "rt-2" + + +def test_revoked_token_disconnects(tmp_path): + clock = Clock() + mb = _mb(tmp_path, [ + httpx.Response(200, json={"access_token": _jwt({"email": "unsubscribe-bot@example.org"}), + "refresh_token": "rt-1", "expires_in": 10}), + httpx.Response(400, json={"error": "invalid_grant"}), + ], clock=clock) + mb.finish(_start_state(mb), "code") + clock.now += 100 + with pytest.raises(oauth.OAuthError): + mb.access_token() + assert not mb.connected() + + +def test_not_connected(tmp_path): + with pytest.raises(oauth.OAuthError): + _mb(tmp_path, []).access_token() + + +def test_app_uses_xoauth2_when_oauth_configured(monkeypatch): + import app as unsub + + calls = {} + + class FakeIMAP: + def __init__(self, host, port): + pass + + def authenticate(self, mech, cb): + calls["auth"] = (mech, cb(b"")) + + def login(self, *a): + calls["login"] = a + + def select(self, folder): + pass + + class FakeOAuth: + def access_token(self): + return "tok" + + monkeypatch.setattr(unsub.imaplib, "IMAP4_SSL", FakeIMAP) + monkeypatch.setattr(unsub, "OAUTH", FakeOAuth()) + monkeypatch.setattr(unsub, "EMAIL_USERNAME", "unsubscribe-bot@example.org") + unsub.connect() + assert calls == {"auth": ("XOAUTH2", oauth.xoauth2_string("unsubscribe-bot@example.org", "tok").encode())} + + +def test_index_shows_connect_when_not_connected(monkeypatch): + import app as unsub + + class FakeOAuth: + def connected(self): + return False + + monkeypatch.setattr(unsub, "OAUTH", FakeOAuth()) + html = unsub.app.test_client().get("/").get_data(as_text=True) + assert 'action="/connect"' in html and 'action="/result"' not in html -- 2.54.0