This commit is contained in:
Executable
+225
@@ -0,0 +1,225 @@
|
||||
#!/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())
|
||||
Reference in New Issue
Block a user