mirror of
https://github.com/chatmail/relay.git
synced 2026-05-11 16:34:39 +00:00
Compare commits
3 Commits
link2xt/py
...
link2xt/on
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
14342383cf | ||
|
|
926de76010 | ||
|
|
ee25d35db1 |
@@ -1,4 +1,6 @@
|
|||||||
import random
|
import random
|
||||||
|
from pathlib import Path
|
||||||
|
import os
|
||||||
import importlib.resources
|
import importlib.resources
|
||||||
import itertools
|
import itertools
|
||||||
from email.parser import BytesParser
|
from email.parser import BytesParser
|
||||||
@@ -57,7 +59,12 @@ def db(tmpdir):
|
|||||||
|
|
||||||
@pytest.fixture
|
@pytest.fixture
|
||||||
def maildata(request):
|
def maildata(request):
|
||||||
datadir = importlib.resources.files(__package__).joinpath("mail-data")
|
try:
|
||||||
|
datadir = importlib.resources.files(__package__).joinpath("mail-data")
|
||||||
|
except TypeError:
|
||||||
|
# in python3.9 or lower, the above doesn't work, so we get datadir this way:
|
||||||
|
datadir = Path(os.getcwd()).joinpath("chatmaild/src/chatmaild/tests/mail-data")
|
||||||
|
|
||||||
assert datadir.exists(), datadir
|
assert datadir.exists(), datadir
|
||||||
|
|
||||||
def maildata(name, from_addr, to_addr):
|
def maildata(name, from_addr, to_addr):
|
||||||
|
|||||||
@@ -5,6 +5,8 @@ import importlib
|
|||||||
import subprocess
|
import subprocess
|
||||||
import datetime
|
import datetime
|
||||||
|
|
||||||
|
from typing import Optional
|
||||||
|
|
||||||
|
|
||||||
class DNS:
|
class DNS:
|
||||||
def __init__(self, out, mail_domain):
|
def __init__(self, out, mail_domain):
|
||||||
@@ -34,7 +36,7 @@ class DNS:
|
|||||||
cmd = "ip a | grep inet6 | grep 'scope global' | sed -e 's#/64 scope global##' | sed -e 's#inet6##'"
|
cmd = "ip a | grep inet6 | grep 'scope global' | sed -e 's#/64 scope global##' | sed -e 's#inet6##'"
|
||||||
return self.shell(cmd).strip()
|
return self.shell(cmd).strip()
|
||||||
|
|
||||||
def get(self, typ: str, domain: str) -> str | None:
|
def get(self, typ: str, domain: str) -> Optional[str]:
|
||||||
"""Get a DNS entry"""
|
"""Get a DNS entry"""
|
||||||
dig_result = self.shell(f"dig -r -q {domain} -t {typ} +short")
|
dig_result = self.shell(f"dig -r -q {domain} -t {typ} +short")
|
||||||
line = dig_result.partition("\n")[0]
|
line = dig_result.partition("\n")[0]
|
||||||
@@ -54,22 +56,25 @@ def show_dns(args, out) -> int:
|
|||||||
ssh = f"ssh root@{mail_domain}"
|
ssh = f"ssh root@{mail_domain}"
|
||||||
dns = DNS(out, mail_domain)
|
dns = DNS(out, mail_domain)
|
||||||
|
|
||||||
def read_dkim_entries(entry):
|
|
||||||
lines = []
|
|
||||||
for line in entry.split("\n"):
|
|
||||||
if line.startswith(";") or not line.strip():
|
|
||||||
continue
|
|
||||||
line = line.replace("\t", " ")
|
|
||||||
lines.append(line)
|
|
||||||
return "\n".join(lines)
|
|
||||||
|
|
||||||
print("Checking your DKIM keys and DNS entries...")
|
print("Checking your DKIM keys and DNS entries...")
|
||||||
try:
|
try:
|
||||||
acme_account_url = out.shell_output(f"{ssh} -- acmetool account-url")
|
acme_account_url = out.shell_output(f"{ssh} -- acmetool account-url")
|
||||||
except subprocess.CalledProcessError:
|
except subprocess.CalledProcessError:
|
||||||
print("Please run `cmdeploy run` first.")
|
print("Please run `cmdeploy run` first.")
|
||||||
return 1
|
return 1
|
||||||
dkim_entry = read_dkim_entries(out.shell_output(f"{ssh} -- opendkim-genzone -F"))
|
|
||||||
|
dkim_selector = "opendkim"
|
||||||
|
dkim_pubkey = out.shell_output(
|
||||||
|
ssh + f" -- openssl rsa -in /etc/dkimkeys/{dkim_selector}.private"
|
||||||
|
" -pubout 2>/dev/null | awk '/-/{next}{printf(\"%s\",$0)}'"
|
||||||
|
)
|
||||||
|
dkim_entry_value = f"v=DKIM1;k=rsa;p={dkim_pubkey};s=email;t=s"
|
||||||
|
dkim_entry_str = ""
|
||||||
|
while len(dkim_entry_value) >= 255:
|
||||||
|
dkim_entry_str += '"' + dkim_entry_value[:255] + '" '
|
||||||
|
dkim_entry_value = dkim_entry_value[255:]
|
||||||
|
dkim_entry_str += '"' + dkim_entry_value + '"'
|
||||||
|
dkim_entry = f"{dkim_selector}._domainkey.{mail_domain}. TXT {dkim_entry_str}"
|
||||||
|
|
||||||
ipv6 = dns.get_ipv6()
|
ipv6 = dns.get_ipv6()
|
||||||
reverse_ipv6 = dns.check_ptr_record(ipv6, mail_domain)
|
reverse_ipv6 = dns.check_ptr_record(ipv6, mail_domain)
|
||||||
@@ -97,7 +102,6 @@ def show_dns(args, out) -> int:
|
|||||||
return 0
|
return 0
|
||||||
except TypeError:
|
except TypeError:
|
||||||
pass
|
pass
|
||||||
started_dkim_parsing = False
|
|
||||||
for line in zonefile.splitlines():
|
for line in zonefile.splitlines():
|
||||||
line = line.format(
|
line = line.format(
|
||||||
acme_account_url=acme_account_url,
|
acme_account_url=acme_account_url,
|
||||||
@@ -124,28 +128,23 @@ def show_dns(args, out) -> int:
|
|||||||
current = dns.get("SRV", domain[:-1])
|
current = dns.get("SRV", domain[:-1])
|
||||||
if current != f"{prio} {weight} {port} {value}":
|
if current != f"{prio} {weight} {port} {value}":
|
||||||
to_print.append(line)
|
to_print.append(line)
|
||||||
if " TXT " in line:
|
if " TXT " in line:
|
||||||
domain, value = line.split(" TXT ")
|
domain, value = line.split(" TXT ")
|
||||||
current = dns.get("TXT", domain.strip()[:-1])
|
current = dns.get("TXT", domain.strip()[:-1])
|
||||||
if domain.startswith("_mta-sts."):
|
if domain.startswith("_mta-sts."):
|
||||||
if current:
|
if current:
|
||||||
if current.split("id=")[0] == value.split("id=")[0]:
|
if current.split("id=")[0] == value.split("id=")[0]:
|
||||||
continue
|
continue
|
||||||
if current != value:
|
|
||||||
|
# TXT records longer than 255 bytes
|
||||||
|
# are split into multiple <character-string>s.
|
||||||
|
# This typically happens with DKIM record
|
||||||
|
# which contains long RSA key.
|
||||||
|
#
|
||||||
|
# Removing `" "` before comparison
|
||||||
|
# to get back a single string.
|
||||||
|
if current.replace('" "', "") != value.replace('" "', ""):
|
||||||
to_print.append(line)
|
to_print.append(line)
|
||||||
if "IN TXT ( " in line:
|
|
||||||
started_dkim_parsing = True
|
|
||||||
dkim_lines = [line]
|
|
||||||
if started_dkim_parsing and line.startswith('"'):
|
|
||||||
dkim_lines.append(" " + line)
|
|
||||||
domain, data = "\n".join(dkim_lines).split(" IN TXT ")
|
|
||||||
current = dns.get("TXT", domain.strip()[:-1])
|
|
||||||
if current:
|
|
||||||
current = "( %s )" % (current.replace('" "', '"\n "'))
|
|
||||||
if current.replace(";", "\\;") != data:
|
|
||||||
to_print.append(dkim_entry)
|
|
||||||
else:
|
|
||||||
to_print.append(dkim_entry)
|
|
||||||
|
|
||||||
exit_code = 0
|
exit_code = 0
|
||||||
if to_print:
|
if to_print:
|
||||||
|
|||||||
Reference in New Issue
Block a user