coredesktop/system/email-watch

#!/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 <email> 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/<domain>.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())