easywks/client.py

643 lines
25 KiB
Python
Executable file

#!/usr/bin/env python3
import abc
import time
from getpass import getpass
from datetime import datetime
import imaplib
import poplib
import smtplib
import subprocess
import urllib.error
import urllib.parse
import urllib.request
import email
from email.mime.application import MIMEApplication
from email.mime.text import MIMEText
from email.parser import BytesParser
from email.utils import format_datetime
from xml.etree import ElementTree
def _get_pgp_message(message: email.message.Message) -> bytes:
pgp = None
for part in message.walk():
if part.is_multipart():
continue
p = part.get_content()
if isinstance(p, str):
p = p.encode()
if b'BEGIN PGP MESSAGE' in p:
if pgp is not None:
raise ValueError('More than one encrypted message part')
pgp = p
if pgp is None:
raise ValueError('No encrypted message part')
return pgp
class MailServerConfig(abc.ABC):
def __init__(self, proto, hostname, port, tls, username, password):
self.proto = proto
self.hostname = hostname
self.port = port
self.tls = tls
self.username = username
self.password = password
def __repr__(self):
proto = self.proto + ('s' if self.tls == 'SSL' else '') + ('+starttls' if self.tls == 'STARTTLS' else '')
return f'{proto}://{self.username}@{self.hostname}:{self.port}'
def __eq__(self, other):
if not isinstance(other, MailServerConfig):
return False
return self.proto == other.proto \
and self.hostname == other.hostname \
and self.port == other.port \
and self.tls == other.tls \
and self.username == other.username \
and self.password == other.password
def __lt__(self, other):
if not isinstance(other, MailServerConfig):
raise TypeError(f'Cannot compare {type(self)} to {type(other)}')
if self == other:
return False
# SMTP not comparable to IMAP or POP3
if (self.proto == 'smtp' or other.proto == 'smtp') and self.proto != other.proto:
raise TypeError(f'Cannot compare {type(self)} to {type(other)}')
# IMAP < POP3
if self.proto == 'imap' and other.proto == 'pop3':
return True
# SSL < STARTTLS < plain
if (self.tls == 'SSL' and other.tls != 'SSL') or (self.tls == 'STARTTLS' and other.tls == 'plain'):
return True
# full email < local part
if '@' in self.username and '@' not in other.username:
return True
def __enter__(self):
pass
def __exit__(self, exc_type, exc_val, exc_tb):
pass
class SmtpServerConfig(MailServerConfig):
def __init__(self, hostname, port, tls, username, password):
super().__init__('smtp', hostname, port, tls, username, password)
self._smtp = None
def __enter__(self):
cls = smtplib.SMTP if self.tls != 'SSL' else smtplib.SMTP_SSL
self._smtp = cls(self.hostname, self.port)
smtp = self._smtp.__enter__()
if self.tls == 'STARTTLS':
smtp.starttls()
smtp.login(self.username, self.password)
return smtp
def __exit__(self, exc_type, exc_val, exc_tb):
ret = self._smtp.__exit__(exc_type, exc_val, exc_tb)
self._smtp = None
return ret
def send_message(self, msg):
if self._smtp is None:
raise RuntimeError('SMTP connection is not established')
self._smtp.send_message(msg)
class IncomingServerConfig(MailServerConfig, abc.ABC):
def __init__(self, proto, hostname, port, tls, username, password):
super().__init__(proto, hostname, port, tls, username, password)
def __enter__(self):
pass
def __exit__(self, exc_type, exc_val, exc_tb):
pass
def get_new_messages(self):
pass
class ImapServerConfig(IncomingServerConfig):
def __init__(self, hostname, port, tls, username, password):
super().__init__('imap', hostname, port, tls, username, password)
self._imap = None
self._uidnext = None
def __enter__(self):
cls = imaplib.IMAP4 if self.tls != 'SSL' else imaplib.IMAP4_SSL
self._imap = cls(self.hostname, self.port)
imap = self._imap.__enter__()
if self.tls == 'STARTTLS':
imap.starttls()
imap.login(self.username, self.password)
imap.select('INBOX', readonly=True)
self._uidnext = int(imap.response('UIDNEXT')[1][0].decode())
return imap
def __exit__(self, exc_type, exc_val, exc_tb):
ret = self._imap.__exit__(exc_type, exc_val, exc_tb)
self._imap = None
self._uidnext = None
return ret
def get_new_messages(self):
if self._imap is None or self._uidnext is None:
raise RuntimeError('IMAP connection is not established')
self._imap.select('INBOX', readonly=True)
u = int(self._imap.response('UIDNEXT')[1][0].decode())
if u > self._uidnext:
messages = self._imap.uid('fetch', f'{self._uidnext}:{u - 1}', '(RFC822)')
for message in messages[1]:
if not isinstance(message, tuple) or b'RFC822' not in message[0]:
# Not an email message
continue
yield message[1]
self._uidnext = u
class Pop3ServerConfig(IncomingServerConfig):
def __init__(self, hostname, port, tls, username, password):
super().__init__('pop3', hostname, port, tls, username, password)
self._pop3: poplib.POP3 = None
self._messagelist = None
def __enter__(self):
cls = poplib.POP3 if self.tls != 'SSL' else poplib.POP3_SSL
self._pop3 = cls(self.hostname, self.port)
if self.tls == 'STARTTLS':
self._pop3.stls()
self._pop3.user(self.username)
self._pop3.pass_(self.password)
self._messagelist = set(self._pop3.list()[1])
return self._pop3
def __exit__(self, exc_type, exc_val, exc_tb):
self._messagelist = None
self._pop3.close()
self._pop3 = None
def get_new_messages(self):
if self._pop3 is None or self._messagelist is None:
raise RuntimeError('POP3 connection is not established')
ml = set(self._pop3.list()[1])
for msg in ml.difference(self._messagelist):
msgid, _ = msg.split(' ', 1)
message = '\r\n'.join(self._pop3.retr(msgid)[1])
yield message[1]
self._messagelist = ml
def default_port(proto: str, tls: str):
if proto == 'imap':
return 993 if tls == 'SSL' else 143
if proto == 'pop3':
return 995 if tls == 'SSL' else 110
if proto == 'smtp':
return 465 if tls == 'SSL' else 587
raise ValueError(f'Unknown protocol: {proto}')
def template_username(template: str, address: str, userinputs):
user, domain = address.split('@', 1)
template = template \
.replace('%EMAILADDRESS%', address) \
.replace('%EMAILLOCALPART%', user) \
.replace('%EMAILDOMAIN%', domain)
for key in userinputs:
if key not in template:
continue
label, value = userinputs[key]
if value is None:
value = getpass(f'Autoconfiguration field "{label}" (will not echo): ')
userinputs[key] = label, value
template = template.replace(key, value)
return template
def parse_xml_mailserver(xmle, address: str, password: str, userinputs):
proto = xmle.get('type')
hostname = xmle.find('./hostname').text
tls = xmle.find('./socketType').text
port = int(xmle.find('./port').text or default_port(proto, tls))
username = template_username(xmle.find('./username').text, address, userinputs)
pw = xmle.find('./password')
if pw:
pw = template_username(pw.text, address, userinputs)
else:
pw = password
if proto == 'smtp':
return SmtpServerConfig(hostname, port, tls, username, pw)
elif proto == 'imap':
return ImapServerConfig(hostname, port, tls, username, pw)
elif proto == 'pop3':
return Pop3ServerConfig(hostname, port, tls, username, pw)
def parse_thunderbird_autoconfig(xml: str, address: str, password: str):
user, domain = address.split('@', 1)
root = ElementTree.fromstring(xml)
userinputs = {}
for ui in root.findall('.//userinput'):
k = ui.get('key')
label = ui.get('label')
userinputs[k] = (label, None)
incoming = root.findall(f"./emailProvider/domain[.='{domain}']/../incomingServer")
outgoing = root.findall(f"./emailProvider/domain[.='{domain}']/../outgoingServer")
iconf = [parse_xml_mailserver(i, address, password, userinputs) for i in incoming]
oconf = [parse_xml_mailserver(o, address, password, userinputs) for o in outgoing]
return iconf, oconf
def tb_wellknown_autoconfig(address: str, password: str):
user, domain = address.split('@', 1)
subdomain = f'autoconfig.{domain}'
sdurl = urllib.parse.urlunsplit(('http', subdomain, 'mail/config-v1.1.xml', f'emailaddress={address}', ''))
mdurl = urllib.parse.urlunsplit(('http', domain, '.well-known/autoconfig/mail/config-v1.1.xml',
f'emailaddress={address}', ''))
try:
with urllib.request.urlopen(sdurl) as sdresponse:
return parse_thunderbird_autoconfig(sdresponse.read().decode(), address, password)
except urllib.error.URLError:
try:
with urllib.request.urlopen(mdurl) as mdresponse:
return parse_thunderbird_autoconfig(mdresponse.read().decode(), address, password)
except urllib.error.URLError:
return None
def tb_ispdb_autoconfig(address: str, password: str):
user, domain = address.split('@', 1)
ispdb = f'https://autoconfig.thunderbird.net/v1.1/{domain}'
try:
with urllib.request.urlopen(ispdb) as response:
return parse_thunderbird_autoconfig(response.read().decode(), address, password)
except urllib.error.URLError:
return None
def manual_config(address: str, password: str):
print('Autoconfiguration has failed. Please enter the mail server settings manually.')
host = input('SMTP hostname: ')
tls = input('TLS ("plain", "SSL" or "STARTLS (default: STARTTLS): ') or 'STARTTLS'
p = default_port('smtp', tls)
port = int(input(f'SMTP port (default: {p}): ') or str(p))
outconfig = SmtpServerConfig(host, port, tls, address, password)
proto = input('Mail Retrieval protocol ("imap" or "pop3", default: "imap"): ') or 'imap'
host = input(f'{proto.upper()} hostname: ')
tls = input('TLS ("plain", "SSL" or "STARTLS (default: SSL): ') or 'SSL'
p = default_port(proto, tls)
port = int(input(f'{proto.upper()} port (default: {p}): ') or str(p))
if proto == 'imap':
inconfig = ImapServerConfig(host, port, tls, address, password)
else:
inconfig = Pop3ServerConfig(host, port, tls, address, password)
return [inconfig], [outconfig]
def parse_dns_srv_record(record, service: str, address: str, password: str):
user, domain = address.split('@', 1)
if service == '_submission':
cls = SmtpServerConfig
elif service == '_imaps' or service == '_imap':
cls = ImapServerConfig
elif service == '_pop3s' or service == '_pop3':
cls = Pop3ServerConfig
else:
raise ValueError(service)
if service.endswith('s'):
tls = ['SSL']
else:
tls = ['STARTTLS', 'plain']
configs = []
for t in tls:
configs.append(cls(record.target, record.port, t, address, password))
configs.append(cls(record.target, record.port, t, user, password))
return configs
def rfc6186_autoconfig(address: str, password: str):
try:
import dns.resolver
import dns.rdtypes.IN.SRV
except ImportError:
print('"dnspython" dependency missing. Skipping RFC 6186 autoconfig')
return None
user, domain = address.split('@', 1)
smtpdomain = f'_submission._tcp.{domain}.'
imapsdomain = f'_imaps._tcp.{domain}.'
imapdomain = f'_imap._tcp.{domain}.'
pop3sdomain = f'_pop3s._tcp.{domain}.'
pop3domain = f'_pop3._tcp.{domain}.'
osrv = []
isrv = []
try:
smtpanswer: dns.resolver.Answer = dns.resolver.resolve(smtpdomain, dns.rdatatype.SRV)
except dns.exception.DNSException:
return None
for srv in smtpanswer:
if srv.rdtype != dns.rdatatype.SRV or srv.rdclass != dns.rdataclass.IN:
continue
osrv.append((smtpanswer.canonical_name.labels[0].decode(), srv))
for sd in [imapsdomain, imapdomain, pop3sdomain, pop3domain]:
try:
answer: dns.resolver.Answer = dns.resolver.resolve(sd, dns.rdatatype.SRV)
for srv in answer:
if srv.rdtype != dns.rdatatype.SRV or srv.rdclass != dns.rdataclass.IN:
continue
if srv.port == 0 and str(srv.target) == '.':
continue
isrv.append((answer.canonical_name.labels[0].decode(), srv))
except dns.exception.DNSException:
continue
incoming = []
outgoing = []
for (iqname, irec) in isrv:
incoming.extend(parse_dns_srv_record(irec, iqname, address, password))
for (oqname, orec) in isrv:
outgoing.extend(parse_dns_srv_record(orec, oqname, address, password))
return incoming, outgoing
def _parse_confirmation_request(address, fingerprint, encrypted):
gpg = subprocess.Popen(['/usr/bin/gpg', '--decrypt'],
stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
gpg.stdin.write(encrypted)
gpg.stdin.close()
gpg.wait()
if gpg.returncode != 0:
raise RuntimeError(f'gpg subprocess returned with non-zero exit code; stderr: {gpg.stderr.read()}')
decrypted = gpg.stdout.read().decode()
rdict = {}
for line in decrypted.splitlines():
if ':' not in line:
continue
key, value = line.split(':', 1)
rdict[key.strip()] = value.strip()
if rdict.get('type', '') != 'confirmation-request':
raise ValueError('Invalid confirmation request: "type" missing or not "confirmation-request"')
if 'sender' not in rdict or 'address' not in rdict or 'fingerprint' not in rdict or 'nonce' not in rdict:
raise ValueError('Invalid confirmation request: a mandatory item is missing from the request')
if rdict['address'] != address:
raise ValueError(f'Confirmation address "{rdict["address"]}" does not match my address "{address}"')
if rdict['fingerprint'].replace(' ', '') != fingerprint.replace(' ', ''):
raise ValueError(
f'Confirmation fingerprint "{rdict["fingerprint"]}" does not match my fingerprint "{fingerprint}"')
print(f'Nonce: {rdict["nonce"]}')
return rdict['sender'], rdict['nonce']
def _create_submission_request(address: str, submission_address: str, fingerprint: str, revoked_fingerprints):
gpg = subprocess.Popen([
'/usr/bin/gpg', '--locate-keys', '--with-colons', submission_address
], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
gpg.wait()
if gpg.returncode != 0:
raise RuntimeError(f'gpg subprocess returned with non-zero exit code; stderr: {gpg.stderr.read()}')
print('Retrieved submission key')
gpg = subprocess.Popen([
'/usr/bin/gpg', '--armor',
'--export-options', 'export-minimal',
'--export', fingerprint
] + revoked_fingerprints, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
gpg.wait()
if gpg.returncode != 0:
raise RuntimeError(f'gpg subprocess returned with non-zero exit code; stderr: {gpg.stderr.read()}')
print('Retrieved key to publish')
pubkey = gpg.stdout.read()
gpg = subprocess.Popen([
'/usr/bin/gpg', '--armor', '--with-colons',
'--encrypt', '--trust-model', 'always',
'--local-user', address,
'--recipient', address,
'--recipient', submission_address
], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
gpg.stdin.write(pubkey)
gpg.stdin.close()
try:
gpg.wait(timeout=2)
except subprocess.TimeoutExpired:
gpg.kill()
print(f'gpg subprocess timed out; stderr: {gpg.stderr.read()}')
raise RuntimeError(f'gpg subprocess timed out; stderr: {gpg.stderr.read()}')
if gpg.returncode != 0:
raise RuntimeError(f'gpg subprocess returned with non-zero exit code; stderr: {gpg.stderr.read()}')
print('Created encrypted message')
encrypted = gpg.stdout.read().decode()
mail = MIMEText(encrypted, _subtype='plain')
mail['Subject'] = 'WKS submission request'
mail['To'] = submission_address
mail['From'] = address
mail['Date'] = format_datetime(datetime.utcnow())
return mail
def _create_confirmation_response(address: str, submission: str, nonce: str, fp: str, content_subtype: str):
response_template = '\r\n'.join([
'type: confirmation-response',
f'sender: {address}',
f'nonce: {nonce}',
''
])
payload = MIMEText(response_template, _subtype='plain')
gpg = subprocess.Popen([
'/usr/bin/gpg', '--armor', '--with-colons',
'--encrypt', '--sign', '--trust-model', 'always',
'--local-user', fp,
'--recipient', address,
'--recipient', submission
], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
gpg.stdin.write(payload.as_string(policy=email.policy.default).encode())
gpg.stdin.close()
gpg.wait()
if gpg.returncode != 0:
raise RuntimeError(f'gpg subprocess returned with non-zero exit code; stderr: {gpg.stderr.read()}')
encrypted = gpg.stdout.read().decode()
mail = MIMEApplication(encrypted, _subtype=content_subtype)
mail['Subject'] = 'WKS confirmation response'
mail['To'] = submission
mail['From'] = address
mail['Date'] = format_datetime(datetime.utcnow())
return mail
def handle_incoming_message(address, fingerprint, rfc822, smtp_config: SmtpServerConfig):
msg: email.message.EmailMessage = BytesParser(policy=email.policy.default).parsebytes(rfc822)
if msg.get('wks-phase', '') == 'done':
pgp = _get_pgp_message(msg)
if pgp is None:
print('WKS key submission successful. Congratulations!')
return True
print('Decrypting WKS response. GnuPG may prompt you for your passphrase.')
gpg = subprocess.Popen(['/usr/bin/gpg', '--batch', '--decrypt', '--skip-verify'],
stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
gpg.stdin.write(pgp)
gpg.stdin.close()
gpg.wait()
if gpg.returncode != 0:
print('It seems WKS submission was successful, however decryption of the submission response failed: ' +
gpg.stderr.read().decode())
return True
decrypted = email.parser.BytesParser(policy=email.policy.default).parsebytes(gpg.stdout.read())
body = decrypted.get_body(preferencelist=('plain',))
if body is None:
print('WKS key submission successful. Congratulations!')
else:
print(body.get_content())
return True
if msg.get('wks-phase', '') == 'error':
body = msg.get_body(preferencelist=('plain',))
if body is None:
print('WKS key submission failed. However, the WKS server did not return an error description.')
else:
print(body.get_content())
return True
for leaf in msg.walk():
if leaf.get_content_type() not in ['application/vnd.gnupg.wkd', 'application/vnd.gnupg.wks']:
continue
print('Received confirmation request')
try:
submission, nonce = _parse_confirmation_request(address, fingerprint, leaf.get_content())
except BaseException as e:
print(f'Parsing failed: {e}')
continue
print('Creating confirmation response. GnuPG may prompt you for your passphrase.')
response = _create_confirmation_response(address, submission, nonce, fingerprint, leaf.get_content_subtype())
print('Sending confirmation response')
with smtp_config:
smtp_config.send_message(response)
print('Awaiting publish response')
return False
def _gpg_get_uid_fp(address: str):
gpg = subprocess.Popen([
'/usr/bin/gpg', '--with-colons', '--list-keys', address
], stdout=subprocess.PIPE)
gpg.wait()
if gpg.returncode != 0:
raise RuntimeError(f'gpg subprocess returned with non-zero exit code; stderr: {gpg.stderr.read()}')
keylist = gpg.stdout.read().decode()
pubs = []
revoked = []
fprs = []
for line in keylist.splitlines():
if line.startswith('pub:'):
pub = line.split(':')[4]
r = line.split(':')[1] == 'r'
pubs.append(pub)
revoked.append(r)
elif line.startswith('fpr:'):
fpr = line.split(':')[9]
fprs.append(fpr)
valid = {next((f for f in fprs if f.endswith(pub))): pub for i, pub in enumerate(pubs) if not revoked[i]}
revoked = {next((f for f in fprs if f.endswith(pub))): pub for i, pub in enumerate(pubs) if revoked[i]}
if len(valid) == 0:
raise ValueError(f'No valid key found for {address}.')
elif len(valid) > 1:
print(f'Found multiple keys for {address}, please choose:')
fpridx = list(valid.keys())
for i, f in enumerate(fpridx, start=1):
print(f'{i}: {f}')
i = int(input('Enter number: ')) - 1
fpr = fpridx[i]
else:
fpr = list(valid.keys())[0]
if len(revoked) > 0:
print(f'There are revoked keys for {address}. Please choose which to upload (separate multiple by spaces): ')
revidx = list(revoked.keys())
for i, f in enumerate(revidx, start=1):
print(f'{i}: {f}')
rids = [int(i)-1 for i in input('Enter number(s): ').split()]
rfprs = [revidx[i] for i in rids]
else:
rfprs = []
return fpr, rfprs
def _get_submission_address(address: str):
_, domain = address.split('@', 1)
advanced_url = f'https://openpgpkey.{domain}/.well-known/openpgpkey/{domain}/submission-address'
direct_url = f'https://{domain}/.well-known/openpgpkey/submission-address'
try:
with urllib.request.urlopen(advanced_url) as response:
return response.read().decode().strip()
except urllib.error.URLError:
pass
with urllib.request.urlopen(direct_url) as response:
return response.read().decode().strip()
def main():
ad = input('Enter email: ')
sa = None
try:
sa = _get_submission_address(ad)
except urllib.error.URLError:
print('No WKS submission address found. Does your provider support WKS?')
exit(1)
fp, rfprs = _gpg_get_uid_fp(ad)
print(f'Chose {fp}')
for rfpr in rfprs:
print(f'Chose revoked key {rfpr}')
pw = getpass('Enter IMAP/POP3/SMTP password (will not echo): ')
for fn in [tb_wellknown_autoconfig, rfc6186_autoconfig, tb_ispdb_autoconfig, manual_config]:
autoconf = fn(ad, pw)
if autoconf is not None:
break
if autoconf is None:
raise RuntimeError('No autoconfig available')
incoming, outgoing = autoconf
# Find the first working server configurations
incoming_server = None
outgoing_server = None
for i in sorted(incoming):
try:
with i:
incoming_server = i
except:
continue
if incoming_server is None:
raise RuntimeError('No working IMAP/POP3 server found through autoconfiguration. Please specify manually.')
print(f'Autoconfigured incoming server: {incoming_server}')
for i in sorted(outgoing):
try:
with i:
outgoing_server = i
except:
continue
if outgoing_server is None:
raise RuntimeError('No working SMTP server found through autoconfiguration. Please specify manually.')
print(f'Autoconfigured outgoing server: {outgoing_server}')
confirm = input('Please confirm: [Y/n] ')
if confirm.lower() not in ['', 'y', 'yes']:
print('Aborted')
exit(1)
with incoming_server:
now = datetime.utcnow()
done = False
request = _create_submission_request(ad, sa, fp, rfprs)
print('Sending submission request')
with outgoing_server:
outgoing_server.send_message(request)
print('Awaiting response')
while not done and (datetime.utcnow() - now).total_seconds() < 300:
time.sleep(5)
for message in incoming_server.get_new_messages():
done = handle_incoming_message(ad, fp, message, outgoing_server)
if done:
break
if __name__ == '__main__':
main()