#!/usr/bin/env python3 """Pause a Core's email in Online Accounts when its password stops working. When your Core password changes (e.g. in Superpanel), email in Online Accounts still has the old one. Online Accounts signs in with it each time the network changes, and Evolution each time it fetches mail, and after a few wrong passwords the mail server's fail2ban blocks this computer. Online Accounts doesn't say when a mail server refuses a password (3.58 reports it as a general failure). It does say when Nextcloud refuses one, and Nextcloud turns off its app passwords when your Core password changes. So when the Core's Nextcloud in Online Accounts needs attention, this signs in to Nextcloud, once, with the password email has (your Core password). If Nextcloud says it's wrong, the mail server will too, so this pauses email (configure.d/lib/goa-account pause): nothing signs in with the old password again. It shows a notification; clicking it runs configure core, where "Your Core password changed" saves the new password and ends the pause. It never signs in to the mail server itself. Run by coredesktop-email-watch.service (see system/email-watch.sh). email-watch --check-password checks the password email has with Nextcloud, as above, for configure core's tests: it exits 0 if it works, 2 if Nextcloud says it's wrong, 1 if it couldn't tell. """ import base64 import glob import importlib.machinery import importlib.util import os import subprocess import sys import threading import urllib.error import urllib.request import gi gi.require_version("Gio", "2.0") from gi.repository import Gio, GLib # noqa: E402 COREDESKTOP = os.path.expanduser("~/.local/share/coredesktop") RUN_IN_TERMINAL = os.path.join(COREDESKTOP, "system/run-in-terminal") CORES = os.path.expanduser("~/.config/coredesktop/cores/*.conf") def load_goa_account(): # No __pycache__ in the CoreDesktop checkout, which update pulls into sys.dont_write_bytecode = True path = os.path.join(COREDESKTOP, "configure.d/lib/goa-account") loader = importlib.machinery.SourceFileLoader("goa_account", path) module = importlib.util.module_from_spec(importlib.util.spec_from_loader("goa_account", loader)) loader.exec_module(module) return module goa = load_goa_account() # Each Core's email and domain, from configure core's cores/.conf def cores(): found = {} for conf in glob.glob(CORES): result = subprocess.run( ["bash", "-c", 'source "$1" && printf "%s\\n%s\\n" "$CORE_EMAIL" "$CORE_DOMAIN"', "-", conf], capture_output=True, text=True) lines = result.stdout.split("\n") if result.returncode == 0 and len(lines) >= 2 and lines[0] and lines[1]: found[lines[0]] = lines[1] return found def nextcloud_url(domain): return f"https://nextcloud.{domain}" # Sign in to the Core's Nextcloud with the password email has, the way # configure core checks a password (configure.d/lib/nextcloud.sh). True if it # works, False if Nextcloud says it's wrong, None if it couldn't tell. def nextcloud_takes(email, domain, password): user = email.rsplit("@", 1)[0] auth = base64.b64encode(f"{user}:{password}".encode()).decode() request = urllib.request.Request( f"{nextcloud_url(domain)}/ocs/v2.php/cloud/user?format=json", headers={"Authorization": f"Basic {auth}", "OCS-APIRequest": "true"}) try: with urllib.request.urlopen(request, timeout=15): return True except urllib.error.HTTPError as error: return False if error.code == 401 else None except OSError: return None def email_password(bus, email): for _, account in goa.find_accounts(bus, "imap_smtp", email): return goa.lookup_credentials("imap_smtp", account["Id"]).get("imap-password") return None def check_password(bus, email): domain = cores().get(email) if domain is None: return None password = email_password(bus, email) if not password: return None return nextcloud_takes(email, domain, password) # Waits for a click in its own thread, so the watching goes on meanwhile def notify(email, domain, paused_now): def run(): if paused_now: body = (f"Your Core password seems to have changed: {email} has the old one. " "So the mail server doesn't block this computer for trying it, it's " "paused. Click to enter your new password.") else: body = (f"{email} is paused because it has an old password. " "Click to enter your new Core password.") result = subprocess.run( ["notify-send", "--wait", "--app-name=CoreDesktop", "--icon=mail-unread", "--action=default=Fix", "--action=fix=Enter new password", f"Email for {domain} is paused", body], capture_output=True, text=True) if result.stdout.strip() in ("default", "fix"): subprocess.run([RUN_IN_TERMINAL, "configure core"]) threading.Thread(target=run, daemon=True).start() # Emails already dealt with since they started failing, so each is checked and # notified about once until it works again handled = set() def check(bus): try: accounts = {(account.get("ProviderType"), account.get("PresentationIdentity")): account for _, account in goa.accounts(bus)} except GLib.Error: return for email, domain in cores().items(): mail = accounts.get(("imap_smtp", email)) if mail is None: continue nextcloud = accounts.get(("owncloud", f"{email.rsplit('@', 1)[0]}@nextcloud.{domain}")) mail_failing = bool(mail.get("AttentionNeeded")) nextcloud_failing = bool(nextcloud and nextcloud.get("AttentionNeeded")) if not mail_failing and not nextcloud_failing: handled.discard(email) continue if email in handled: continue try: if goa.paused(bus, "imap_smtp", email): handled.add(email) notify(email, domain, False) continue # Nextcloud may only have had an app password revoked, and email may # still be marked from a pause whose new password hasn't been # checked yet, so check the password email has before pausing it password = email_password(bus, email) if not password: continue # e.g. the keyring is locked; try again later works = nextcloud_takes(email, domain, password) if works is None: continue # try again when the network changes if works: handled.add(email) continue goa.pause(bus, "imap_smtp", email) except (GLib.Error, LookupError): continue handled.add(email) notify(email, domain, True) def watch(): bus = Gio.bus_get_sync(Gio.BusType.SESSION, None) pending = [] # Signals come in bursts; check once they've settled def queue_check(*_): if not pending: pending.append(GLib.timeout_add_seconds(2, run_check)) def run_check(): pending.clear() check(bus) return GLib.SOURCE_REMOVE account = "org.gnome.OnlineAccounts.Account" bus.signal_subscribe(goa.BUS_NAME, "org.freedesktop.DBus.Properties", "PropertiesChanged", None, account, Gio.DBusSignalFlags.NONE, queue_check) bus.signal_subscribe(goa.BUS_NAME, "org.freedesktop.DBus.ObjectManager", "InterfacesAdded", None, None, Gio.DBusSignalFlags.NONE, queue_check) # Online Accounts checks every account when it starts, e.g. at login, and # when the network changes bus.signal_subscribe("org.freedesktop.DBus", "org.freedesktop.DBus", "NameOwnerChanged", "/org/freedesktop/DBus", goa.BUS_NAME, Gio.DBusSignalFlags.NONE, queue_check) Gio.NetworkMonitor.get_default().connect("network-changed", queue_check) queue_check() GLib.MainLoop().run() def main(): if len(sys.argv) == 3 and sys.argv[1] == "--check-password": bus = Gio.bus_get_sync(Gio.BusType.SESSION, None) try: works = check_password(bus, sys.argv[2]) except (GLib.Error, LookupError): works = None return {True: 0, False: 2, None: 1}[works] if len(sys.argv) == 1: watch() return 0 print(__doc__.strip().split("\n\n")[-1], file=sys.stderr) return 2 if __name__ == "__main__": sys.exit(main())