#
# Copyright (c) 2024 YunoHost Contributors
#
# This file is part of YunoHost (see https://yunohost.org)
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as
# published by the Free Software Foundation, either version 3 of the
# License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
import os
import re
import time
from difflib import SequenceMatcher
from collections import OrderedDict
from moulinette import m18n, Moulinette
from moulinette.utils.log import getActionLogger
from moulinette.utils.filesystem import read_file, write_to_file, read_toml, mkdir
from yunohost.domain import (
_assert_domain_exists,
domain_config_get,
_get_domain_settings,
_set_domain_settings,
_list_subdomains_of,
_get_parent_domain_of,
)
from yunohost.utils.dns import dig, is_yunohost_dyndns_domain, is_special_use_tld
from yunohost.utils.error import YunohostValidationError, YunohostError
from yunohost.utils.network import get_public_ip
from yunohost.settings import settings_get
from yunohost.log import is_unit_operation
from yunohost.hook import hook_callback
logger = getActionLogger("yunohost.domain")
DOMAIN_REGISTRAR_LIST_PATH = "/usr/share/yunohost/registrar_list.toml"
def domain_dns_suggest(domain):
"""
Generate DNS configuration for a domain
Keyword argument:
domain -- Domain name
"""
if is_special_use_tld(domain):
return m18n.n("domain_dns_conf_special_use_tld")
_assert_domain_exists(domain)
dns_conf = _build_dns_conf(domain)
result = ""
if dns_conf["basic"]:
result += "; Basic ipv4/ipv6 records"
for record in dns_conf["basic"]:
result += "\n{name} {ttl} IN {type} {value}".format(**record)
if dns_conf["mail"]:
result += "\n\n"
result += "; Mail"
for record in dns_conf["mail"]:
result += "\n{name} {ttl} IN {type} {value}".format(**record)
result += "\n\n"
if dns_conf["xmpp"]:
result += "\n\n"
result += "; XMPP"
for record in dns_conf["xmpp"]:
result += "\n{name} {ttl} IN {type} {value}".format(**record)
if dns_conf["extra"]:
result += "\n\n"
result += "; Extra"
for record in dns_conf["extra"]:
result += "\n{name} {ttl} IN {type} {value}".format(**record)
for name, record_list in dns_conf.items():
if name not in ("basic", "xmpp", "mail", "extra") and record_list:
result += "\n\n"
result += "; " + name
for record in record_list:
result += "\n{name} {ttl} IN {type} {value}".format(**record)
if Moulinette.interface.type == "cli":
# FIXME Update this to point to our "dns push" doc
logger.info(m18n.n("domain_dns_conf_is_just_a_recommendation"))
return result
def _build_dns_conf(base_domain, include_empty_AAAA_if_no_ipv6=False):
"""
Internal function that will returns a data structure containing the needed
information to generate/adapt the dns configuration
Arguments:
domains -- List of a domain and its subdomains
The returned datastructure will have the following form:
{
"basic": [
# if ipv4 available
{"type": "A", "name": "@", "value": "123.123.123.123", "ttl": 3600},
# if ipv6 available
{"type": "AAAA", "name": "@", "value": "valid-ipv6", "ttl": 3600},
],
"xmpp": [
{"type": "SRV", "name": "_xmpp-client._tcp", "value": "0 5 5222 domain.tld.", "ttl": 3600},
{"type": "SRV", "name": "_xmpp-server._tcp", "value": "0 5 5269 domain.tld.", "ttl": 3600},
{"type": "CNAME", "name": "muc", "value": "@", "ttl": 3600},
{"type": "CNAME", "name": "pubsub", "value": "@", "ttl": 3600},
{"type": "CNAME", "name": "vjud", "value": "@", "ttl": 3600}
{"type": "CNAME", "name": "xmpp-upload", "value": "@", "ttl": 3600}
],
"mail": [
{"type": "MX", "name": "@", "value": "10 domain.tld.", "ttl": 3600},
{"type": "TXT", "name": "@", "value": "\"v=spf1 a mx ip4:123.123.123.123 ipv6:valid-ipv6 -all\"", "ttl": 3600 },
{"type": "TXT", "name": "mail._domainkey", "value": "\"v=DKIM1; k=rsa; p=some-super-long-key\"", "ttl": 3600},
{"type": "TXT", "name": "_dmarc", "value": "\"v=DMARC1; p=none\"", "ttl": 3600}
],
"extra": [
# if ipv4 available
{"type": "A", "name": "*", "value": "123.123.123.123", "ttl": 3600},
# if ipv6 available
{"type": "AAAA", "name": "*", "value": "valid-ipv6", "ttl": 3600},
{"type": "CAA", "name": "@", "value": "0 issue \"letsencrypt.org\"", "ttl": 3600},
],
"example_of_a_custom_rule": [
{"type": "SRV", "name": "_matrix", "value": "domain.tld.", "ttl": 3600}
],
}
"""
basic = []
mail = []
xmpp = []
extra = []
ipv4 = get_public_ip()
ipv6 = get_public_ip(6)
# If this is a ynh_dyndns_domain, we're not gonna include all the subdomains in the conf
# Because dynette only accept a specific list of name/type
# And the wildcard */A already covers the bulk of use cases
if is_yunohost_dyndns_domain(base_domain):
subdomains = []
else:
subdomains = _list_subdomains_of(base_domain)
domains_settings = {
domain: domain_config_get(domain, export=True)
for domain in [base_domain] + subdomains
}
base_dns_zone = _get_dns_zone_for_domain(base_domain)
for domain, settings in domains_settings.items():
# Domain # Base DNS zone # Basename # Suffix #
# ------------------ # ----------------- # --------- # -------- #
# domain.tld # domain.tld # @ # #
# sub.domain.tld # domain.tld # sub # .sub #
# foo.sub.domain.tld # domain.tld # foo.sub # .foo.sub #
# sub.domain.tld # sub.domain.tld # @ # #
# foo.sub.domain.tld # sub.domain.tld # foo # .foo #
basename = _get_relative_name_for_dns_zone(domain, base_dns_zone)
suffix = f".{basename}" if basename != "@" else ""
# ttl = settings["ttl"]
ttl = 3600
###########################
# Basic ipv4/ipv6 records #
###########################
if ipv4 and settings_get("misc.network.dns_exposure") in ["both", "ipv4"]:
basic.append([basename, ttl, "A", ipv4])
if ipv6:
basic.append([basename, ttl, "AAAA", ipv6])
elif include_empty_AAAA_if_no_ipv6:
basic.append([basename, ttl, "AAAA", None])
#########
# Email #
#########
if settings["mail_in"]:
mail.append([basename, ttl, "MX", f"10 {domain}."])
if settings["mail_out"]:
mail.append([basename, ttl, "TXT", '"v=spf1 a mx -all"'])
# DKIM/DMARC record
dkim_host, dkim_publickey = _get_DKIM(domain)
if dkim_host:
mail += [
[f"{dkim_host}{suffix}", ttl, "TXT", dkim_publickey],
[f"_dmarc{suffix}", ttl, "TXT", '"v=DMARC1; p=none"'],
]
########
# XMPP #
########
if settings["xmpp"]:
xmpp += [
[
f"_xmpp-client._tcp{suffix}",
ttl,
"SRV",
f"0 5 5222 {domain}.",
],
[
f"_xmpp-server._tcp{suffix}",
ttl,
"SRV",
f"0 5 5269 {domain}.",
],
[f"muc{suffix}", ttl, "CNAME", f"{domain}."],
[f"pubsub{suffix}", ttl, "CNAME", f"{domain}."],
[f"vjud{suffix}", ttl, "CNAME", f"{domain}."],
[f"xmpp-upload{suffix}", ttl, "CNAME", f"{domain}."],
]
#########
# Extra #
#########
# Only recommend wildcard and CAA for the top level
if domain == base_domain:
if ipv4 and settings_get("misc.network.dns_exposure") in ["both", "ipv4"]:
extra.append([f"*{suffix}", ttl, "A", ipv4])
if ipv6:
extra.append([f"*{suffix}", ttl, "AAAA", ipv6])
elif include_empty_AAAA_if_no_ipv6:
extra.append([f"*{suffix}", ttl, "AAAA", None])
extra.append([basename, ttl, "CAA", '0 issue "letsencrypt.org"'])
####################
# Standard records #
####################
records = {
"basic": [
{"name": name, "ttl": ttl_, "type": type_, "value": value}
for name, ttl_, type_, value in basic
],
"xmpp": [
{"name": name, "ttl": ttl_, "type": type_, "value": value}
for name, ttl_, type_, value in xmpp
],
"mail": [
{"name": name, "ttl": ttl_, "type": type_, "value": value}
for name, ttl_, type_, value in mail
],
"extra": [
{"name": name, "ttl": ttl_, "type": type_, "value": value}
for name, ttl_, type_, value in extra
],
}
##################
# Custom records #
##################
# Defined by custom hooks ships in apps for example ...
# FIXME : this ain't practical for apps that may want to add
# custom dns records for a subdomain ... there's no easy way for
# an app to compare the base domain is the parent of the subdomain ?
# (On the other hand, in sep 2021, it looks like no app is using
# this mechanism...)
hook_results = hook_callback("custom_dns_rules", args=[base_domain])
for hook_name, results in hook_results.items():
#
# There can be multiple results per hook name, so results look like
# {'/some/path/to/hook1':
# { 'state': 'succeed',
# 'stdreturn': [{'type': 'SRV',
# 'name': 'stuff.foo.bar.',
# 'value': 'yoloswag',
# 'ttl': 3600}]
# },
# '/some/path/to/hook2':
# { ... },
# [...]
#
# Loop over the sub-results
custom_records = [
v["stdreturn"] for v in results.values() if v and v["stdreturn"]
]
records[hook_name] = []
for record_list in custom_records:
# Check that record_list is indeed a list of dict
# with the required keys
if (
not isinstance(record_list, list)
or any(not isinstance(record, dict) for record in record_list)
or any(
key not in record
for record in record_list
for key in ["name", "ttl", "type", "value"]
)
):
# Display an error, mainly for app packagers trying to implement a hook
logger.warning(
"Ignored custom record from hook '%s' because the data is not a *list* of dict with keys name, ttl, type and value. Raw data : %s"
% (hook_name, record_list)
)
continue
records[hook_name].extend(record_list)
return records
def _get_DKIM(domain):
DKIM_file = f"/etc/dkim/{domain}.mail.txt"
if not os.path.isfile(DKIM_file):
return (None, None)
with open(DKIM_file) as f:
dkim_content = f.read()
# Gotta manage two formats :
#
# Legacy
# -----
#
# mail._domainkey IN TXT ( "v=DKIM1; k=rsa; "
# "p=<theDKIMpublicKey>" )
#
# New
# ------
#
# mail._domainkey IN TXT ( "v=DKIM1; h=sha256; k=rsa; "
# "p=<theDKIMpublicKey>" )
is_legacy_format = " h=sha256; " not in dkim_content
# Legacy DKIM format
if is_legacy_format:
dkim = re.match(
(
r"^(?P<host>[a-z_\-\.]+)[\s]+([0-9]+[\s]+)?IN[\s]+TXT[\s]+"
r'[^"]*"v=(?P<v>[^";]+);'
r'[\s"]*k=(?P<k>[^";]+);'
r'[\s"]*p=(?P<p>[^";]+)'
),
dkim_content,
re.M | re.S,
)
else:
dkim = re.match(
(
r"^(?P<host>[a-z_\-\.]+)[\s]+([0-9]+[\s]+)?IN[\s]+TXT[\s]+"
r'[^"]*"v=(?P<v>[^";]+);'
r'[\s"]*h=(?P<h>[^";]+);'
r'[\s"]*k=(?P<k>[^";]+);'
r'[\s"]*p=(?P<p>[^";]+)'
),
dkim_content,
re.M | re.S,
)
if not dkim:
return (None, None)
if is_legacy_format:
return (
dkim.group("host"),
'"v={v}; k={k}; p={p}"'.format(
v=dkim.group("v"), k=dkim.group("k"), p=dkim.group("p")
),
)
else:
return (
dkim.group("host"),
'"v={v}; h={h}; k={k}; p={p}"'.format(
v=dkim.group("v"),
h=dkim.group("h"),
k=dkim.group("k"),
p=dkim.group("p"),
),
)
def _get_dns_zone_for_domain(domain):
"""
Get the DNS zone of a domain
Keyword arguments:
domain -- The domain name
"""
# First, check if domain is a nohost.me / noho.st / ynh.fr
# This is mainly meant to speed up things for "dyndns update"
# ... otherwise we end up constantly doing a bunch of dig requests
if is_yunohost_dyndns_domain(domain):
# Keep only foo.nohost.me even if we have subsub.sub.foo.nohost.me
return ".".join(domain.rsplit(".", 3)[-3:])
# Same thing with .local, .test, ... domains
if is_special_use_tld(domain):
# Keep only foo.local even if we have subsub.sub.foo.local
return ".".join(domain.rsplit(".", 2)[-2:])
# Check cache
cache_folder = "/var/cache/yunohost/dns_zones"
cache_file = f"{cache_folder}/{domain}"
cache_duration = 3600 # one hour
if (
os.path.exists(cache_file)
and abs(os.path.getctime(cache_file) - time.time()) < cache_duration
):
dns_zone = read_file(cache_file).strip()
if dns_zone:
return dns_zone
# Check cache for parent domain
# This is another strick to try to prevent this function from being
# a bottleneck on system with 1 main domain + 10ish subdomains
# when building the dns conf for the main domain (which will call domain_config_get, etc...)
parent_domain = _get_parent_domain_of(domain)
if parent_domain:
parent_cache_file = f"{cache_folder}/{parent_domain}"
if (
os.path.exists(parent_cache_file)
and abs(os.path.getctime(parent_cache_file) - time.time()) < cache_duration
):
dns_zone = read_file(parent_cache_file).strip()
if dns_zone:
return dns_zone
# For foo.bar.baz.gni we want to scan all the parent domains
# (including the domain itself)
# foo.bar.baz.gni
# bar.baz.gni
# baz.gni
# gni
# Until we find the first one that has a NS record
parent_list = [domain.split(".", i)[-1] for i, _ in enumerate(domain.split("."))]
# We don't wan't to do A NS request on the tld
for parent in parent_list[0:-1]:
# Check if there's a NS record for that domain
answer = dig(parent, rdtype="NS", full_answers=True, resolvers="force_external")
if answer[0] != "ok":
# Some domains have a SOA configured but NO NS record !!!
# See https://github.com/YunoHost/issues/issues/1980
answer = dig(
parent, rdtype="SOA", full_answers=True, resolvers="force_external"
)
if answer[0] == "ok":
mkdir(cache_folder, parents=True, force=True)
write_to_file(cache_file, parent)
return parent
if len(parent_list) >= 2:
zone = parent_list[-2]
else:
zone = parent_list[-1]
logger.warning(
f"Could not identify correctly the dns zone for domain {domain}, returning {zone}"
)
return zone
def _get_relative_name_for_dns_zone(domain, base_dns_zone):
# Strip the base dns zone name from a domain such that it's suitable for DNS manipulation relative to a defined zone
# For example, assuming base_dns_zone is "example.tld":
# example.tld -> @
# foo.example.tld -> foo
# .foo.example.tld -> foo
# bar.foo.example.tld -> bar.foo
return (
re.sub(r"\.?" + base_dns_zone.replace(".", r"\.") + "$", "", domain.strip("."))
or "@"
)
def _get_registrar_config_section(domain):
from lexicon.providers.auto import _relevant_provider_for_domain
registrar_infos = {
"name": m18n.n(
"registrar_infos"
), # This is meant to name the config panel section, for proper display in the webadmin
}
dns_zone = _get_dns_zone_for_domain(domain)
# If parent domain exists in yunohost
parent_domain = _get_parent_domain_of(domain, topest=True)
if parent_domain:
# Dirty hack to have a link on the webadmin
if Moulinette.interface.type == "api":
parent_domain_link = f"[{parent_domain}](#/domains/{parent_domain}/dns)"
else:
parent_domain_link = parent_domain
registrar_infos["registrar"] = OrderedDict(
{
"type": "alert",
"style": "info",
"ask": m18n.n(
"domain_dns_registrar_managed_in_parent_domain",
parent_domain=parent_domain,
parent_domain_link=parent_domain_link,
),
"value": "parent_domain",
}
)
return OrderedDict(registrar_infos)
# TODO big project, integrate yunohost's dynette as a registrar-like provider
# TODO big project, integrate other dyndns providers such as netlib.re, or cf the list of dyndns providers supported by cloudron...
if is_yunohost_dyndns_domain(dns_zone):
registrar_infos["registrar"] = OrderedDict(
{
"type": "alert",
"style": "success",
"ask": m18n.n("domain_dns_registrar_yunohost"),
"value": "yunohost",
}
)
registrar_infos["recovery_password"] = OrderedDict(
{
"type": "password",
"ask": m18n.n("ask_dyndns_recovery_password"),
"default": "",
}
)
return OrderedDict(registrar_infos)
elif is_special_use_tld(dns_zone):
registrar_infos["registrar"] = OrderedDict(
{
"type": "alert",
"style": "info",
"ask": m18n.n("domain_dns_conf_special_use_tld"),
"value": None,
}
)
try:
registrar = _relevant_provider_for_domain(dns_zone)[0]
except ValueError:
registrar_infos["registrar"] = OrderedDict(
{
"type": "alert",
"style": "warning",
"ask": m18n.n("domain_dns_registrar_not_supported"),
"value": None,
}
)
else:
registrar_infos["registrar"] = OrderedDict(
{
"type": "alert",
"style": "info",
"ask": m18n.n("domain_dns_registrar_supported", registrar=registrar),
"value": registrar,
}
)
TESTED_REGISTRARS = ["ovh", "gandi"]
if registrar not in TESTED_REGISTRARS:
registrar_infos["experimental_disclaimer"] = OrderedDict(
{
"type": "alert",
"style": "danger",
"ask": m18n.n(
"domain_dns_registrar_experimental", registrar=registrar
),
}
)
# TODO : add a help tip with the link to the registar's API doc (c.f. Lexicon's README)
registrar_list = read_toml(DOMAIN_REGISTRAR_LIST_PATH)
registrar_credentials = registrar_list.get(registrar)
if registrar_credentials is None:
logger.warning(
f"Registrar {registrar} unknown / Should be added to YunoHost's registrar_list.toml by the development team!"
)
registrar_credentials = {}
for credential, infos in registrar_credentials.items():
infos["default"] = infos.get("default", "")
infos["optional"] = infos.get("optional", "False")
registrar_infos.update(registrar_credentials)
return OrderedDict(registrar_infos)
def _get_registar_settings(domain):
_assert_domain_exists(domain)
settings = domain_config_get(domain, key="dns.registrar", export=True)
registrar = settings.pop("registrar")
if "experimental_disclaimer" in settings:
settings.pop("experimental_disclaimer")
return registrar, settings
@is_unit_operation()
def domain_dns_push(operation_logger, domain, dry_run=False, force=False, purge=False):
"""
Send DNS records to the previously-configured registrar of the domain.
"""
from lexicon.client import Client as LexiconClient
from lexicon.config import ConfigResolver as LexiconConfigResolver
registrar, registrar_credentials = _get_registar_settings(domain)
_assert_domain_exists(domain)
if is_special_use_tld(domain):
logger.info(m18n.n("domain_dns_conf_special_use_tld"))
return {}
if not registrar or registrar == "None": # yes it's None as a string
raise YunohostValidationError("domain_dns_push_not_applicable", domain=domain)
# FIXME: in the future, properly unify this with yunohost dyndns update
if registrar == "yunohost":
from yunohost.dyndns import dyndns_update
dyndns_update(domain=domain, force=force)
return {}
if registrar == "parent_domain":
parent_domain = _get_parent_domain_of(domain, topest=True)
registrar, registrar_credentials = _get_registar_settings(parent_domain)
if any(registrar_credentials.values()):
raise YunohostValidationError(
"domain_dns_push_managed_in_parent_domain",
domain=domain,
parent_domain=parent_domain,
)
else:
new_parent_domain = ".".join(parent_domain.split(".")[-3:])
registrar, registrar_credentials = _get_registar_settings(new_parent_domain)
if registrar == "yunohost":
raise YunohostValidationError(
"domain_dns_push_managed_in_parent_domain",
domain=domain,
parent_domain=new_parent_domain,
)
else:
raise YunohostValidationError(
"domain_registrar_is_not_configured", domain=parent_domain
)
if not all(registrar_credentials.values()):
raise YunohostValidationError(
"domain_registrar_is_not_configured", domain=domain
)
base_dns_zone = _get_dns_zone_for_domain(domain)
# Convert the generated conf into a format that matches what we'll fetch using the API
# Makes it easier to compare "wanted records" with "current records on remote"
wanted_records = []
for records in _build_dns_conf(domain).values():
for record in records:
# Make sure the name is a FQDN
name = (
f"{record['name']}.{base_dns_zone}"
if record["name"] != "@"
else base_dns_zone
)
type_ = record["type"]
content = record["value"]
# Make sure the content is also a FQDN (with trailing . ?)
if content == "@" and record["type"] == "CNAME":
content = base_dns_zone + "."
wanted_records.append(
{"name": name, "type": type_, "ttl": record["ttl"], "content": content}
)
# FIXME Lexicon does not support CAA records
# See https://github.com/AnalogJ/lexicon/issues/282 and https://github.com/AnalogJ/lexicon/pull/371
# They say it's trivial to implement it!
# And yet, it is still not done/merged
# Update by Aleks: it works - at least with Gandi ?!
# wanted_records = [record for record in wanted_records if record["type"] != "CAA"]
if purge:
wanted_records = []
force = True
# Construct the base data structure to use lexicon's API.
base_config = {
"provider_name": registrar,
"domain": base_dns_zone,
registrar: registrar_credentials,
}
# Ugly hack to be able to fetch all record types at once:
# we initialize a LexiconClient with a dummy type "all"
# (which lexicon doesnt actually understands)
# then trigger ourselves the authentication + list_records
# instead of calling .execute()
query = (
LexiconConfigResolver()
.with_dict(dict_object=base_config)
.with_dict(dict_object={"action": "list", "type": "all"})
)
client = LexiconClient(query)
try:
client.provider.authenticate()
except Exception as e:
raise YunohostValidationError(
"domain_dns_push_failed_to_authenticate", domain=domain, error=str(e)
)
try:
current_records = client.provider.list_records()
except Exception as e:
raise YunohostError("domain_dns_push_failed_to_list", error=str(e))
managed_dns_records_hashes = _get_managed_dns_records_hashes(domain)
# Keep only records for relevant types: A, AAAA, MX, TXT, CNAME, SRV
relevant_types = ["A", "AAAA", "MX", "TXT", "CNAME", "SRV", "CAA"]
current_records = [r for r in current_records if r["type"] in relevant_types]
# Ignore records which are for a higher-level domain
# i.e. we don't care about the records for domain.tld when pushing yuno.domain.tld
current_records = [
r
for r in current_records
if r["name"].endswith(f".{domain}") or r["name"] == domain
]
for record in current_records:
# Try to get rid of weird stuff like ".domain.tld" or "@.domain.tld"
record["name"] = record["name"].strip("@").strip(".")
# Some API return '@' in content and we shall convert it to absolute/fqdn
record["content"] = (
record["content"]
.replace("@.", base_dns_zone + ".")
.replace("@", base_dns_zone + ".")
)
if record["type"] == "TXT":
if not record["content"].startswith('"'):
record["content"] = '"' + record["content"]
if not record["content"].endswith('"'):
record["content"] = record["content"] + '"'
# Check if this record was previously set by YunoHost
record["managed_by_yunohost"] = (
_hash_dns_record(record) in managed_dns_records_hashes
)
# Step 0 : Get the list of unique (type, name)
# And compare the current and wanted records
#
# i.e. we want this kind of stuff:
# wanted current
# (A, .domain.tld) 1.2.3.4 1.2.3.4
# (A, www.domain.tld) 1.2.3.4 5.6.7.8
# (A, foobar.domain.tld) 1.2.3.4
# (AAAA, .domain.tld) 2001::abcd
# (MX, .domain.tld) 10 domain.tld [10 mx1.ovh.net, 20 mx2.ovh.net]
# (TXT, .domain.tld) "v=spf1 ..." ["v=spf1", "foobar"]
# (SRV, .domain.tld) 0 5 5269 domain.tld
changes = {"delete": [], "update": [], "create": [], "unchanged": []}
type_and_names = sorted(
{(r["type"], r["name"]) for r in current_records + wanted_records}
)
comparison = {
type_and_name: {"current": [], "wanted": []} for type_and_name in type_and_names
}
for record in current_records:
comparison[(record["type"], record["name"])]["current"].append(record)
for record in wanted_records:
comparison[(record["type"], record["name"])]["wanted"].append(record)
for type_and_name, records in comparison.items():
#
# Step 1 : compute a first "diff" where we remove records which are the same on both sides
#
wanted_contents = [r["content"] for r in records["wanted"]]
current_contents = [r["content"] for r in records["current"]]
current = [r for r in records["current"] if r["content"] not in wanted_contents]
wanted = [r for r in records["wanted"] if r["content"] not in current_contents]
#
# Step 2 : simple case: 0 record on one side, 0 on the other
# -> either nothing do (0/0) or creations (0/N) or deletions (N/0)
#
if len(current) == 0 and len(wanted) == 0:
# No diff, nothing to do
changes["unchanged"].extend(records["current"])
continue
elif len(wanted) == 0:
changes["delete"].extend(current)
continue
elif len(current) == 0:
changes["create"].extend(wanted)
continue
#
# Step 3 : N record on one side, M on the other
#
# Fuzzy matching strategy:
# For each wanted record, try to find a current record which looks like the wanted one
# -> if found, trigger an update
# -> if no match found, trigger a create
#
for record in wanted:
def likeliness(r):
# We compute this only on the first 100 chars, to have a high value even for completely different DKIM keys
return SequenceMatcher(
None, r["content"][:100], record["content"][:100]
).ratio()
matches = sorted(current, key=lambda r: likeliness(r), reverse=True)
if matches and likeliness(matches[0]) > 0.50:
match = matches[0]
# Remove the match from 'current' so that it's not added to the removed stuff later
current.remove(match)
match["old_content"] = match["content"]
match["content"] = record["content"]
changes["update"].append(match)
else:
changes["create"].append(record)
#
# For all other remaining current records:
# -> trigger deletions
#
for record in current:
changes["delete"].append(record)
def human_readable_record(action, record):
name = record["name"]
name = _get_relative_name_for_dns_zone(record["name"], base_dns_zone)
name = name[:20]
t = record["type"]
if not force and action in ["update", "delete"]:
ignored = (
""
if record["managed_by_yunohost"]
else "(ignored, won't be changed by Yunohost unless forced)"
)
else:
ignored = ""
if action == "create":
new_content = record.get("content", "(None)")[:30]
return f"{name:>20} [{t:^5}] {new_content:^30} {ignored}"
elif action == "update":
old_content = record.get("old_content", "(None)")[:30]
new_content = record.get("content", "(None)")[:30]
return (
f"{name:>20} [{t:^5}] {old_content:^30} -> {new_content:^30} {ignored}"
)
elif action == "unchanged":
old_content = record.get("content", "(None)")[:30]
return f"{name:>20} [{t:^5}] {old_content:^30}"
else:
old_content = record.get("content", "(None)")[:30]
return f"{name:>20} [{t:^5}] {old_content:^30} {ignored}"
if dry_run:
if Moulinette.interface.type == "api":
for records in changes.values():
for record in records:
record["name"] = _get_relative_name_for_dns_zone(
record["name"], base_dns_zone
)
return changes
else:
out = {"delete": [], "create": [], "update": [], "unchanged": []}
for action in ["delete", "create", "update", "unchanged"]:
for record in changes[action]:
out[action].append(human_readable_record(action, record))
return out
# If --force ain't used, we won't delete/update records not managed by yunohost
if not force:
for action in ["delete", "update"]:
changes[action] = [r for r in changes[action] if r["managed_by_yunohost"]]
def progress(info=""):
progress.nb += 1
width = 20
bar = int(progress.nb * width / progress.total)
bar = "[" + "#" * bar + "." * (width - bar) + "]"
if info:
bar += " > " + info
if progress.old == bar:
return
progress.old = bar
logger.info(bar)
progress.nb = 0
progress.old = ""
progress.total = len(changes["delete"] + changes["create"] + changes["update"])
if progress.total == 0:
logger.success(m18n.n("domain_dns_push_already_up_to_date"))
return {}
#
# Actually push the records
#
operation_logger.start()
logger.info(m18n.n("domain_dns_pushing"))
new_managed_dns_records_hashes = [_hash_dns_record(r) for r in changes["unchanged"]]
results = {"warnings": [], "errors": []}
for action in ["delete", "create", "update"]:
for record in changes[action]:
relative_name = _get_relative_name_for_dns_zone(
record["name"], base_dns_zone
)
progress(
f"{action} {record['type']:^5} / {relative_name}"
) # FIXME: i18n but meh
# Apparently Lexicon yields us some 'id' during fetch
# But wants 'identifier' during push ...
if "id" in record:
record["identifier"] = record["id"]
del record["id"]
if registrar == "godaddy":
if record["name"] == base_dns_zone:
record["name"] = "@." + record["name"]
if record["type"] in ["MX", "SRV", "CAA"]:
logger.warning(
f"Pushing {record['type']} records is not properly supported by Lexicon/Godaddy."
)
results["warnings"].append(
f"Pushing {record['type']} records is not properly supported by Lexicon/Godaddy."
)
continue
elif registrar == "gandi":
if record["name"] == base_dns_zone:
record["name"] = "@." + record["name"]
record["action"] = action
query = (
LexiconConfigResolver()
.with_dict(dict_object=base_config)
.with_dict(dict_object=record)
)
try:
result = LexiconClient(query).execute()
except Exception as e:
msg = m18n.n(
"domain_dns_push_record_failed",
action=action,
type=record["type"],
name=record["name"],
error=str(e),
)
logger.error(msg)
results["errors"].append(msg)
else:
if result:
new_managed_dns_records_hashes.append(_hash_dns_record(record))
else:
msg = m18n.n(
"domain_dns_push_record_failed",
action=action,
type=record["type"],
name=record["name"],
error="unkonwn error?",
)
logger.error(msg)
results["errors"].append(msg)
_set_managed_dns_records_hashes(domain, new_managed_dns_records_hashes)
# Everything succeeded
if len(results["errors"]) + len(results["warnings"]) == 0:
logger.success(m18n.n("domain_dns_push_success"))
return {}
# Everything failed
elif len(results["errors"]) + len(results["warnings"]) == progress.total:
logger.error(m18n.n("domain_dns_push_failed"))
else:
logger.warning(m18n.n("domain_dns_push_partial_failure"))
return results
def _get_managed_dns_records_hashes(domain: str) -> list:
return _get_domain_settings(domain).get("managed_dns_records_hashes", [])
def _set_managed_dns_records_hashes(domain: str, hashes: list) -> None:
settings = _get_domain_settings(domain)
settings["managed_dns_records_hashes"] = hashes or []
_set_domain_settings(domain, settings)
def _hash_dns_record(record: dict) -> int:
fields = ["name", "type", "content"]
record_ = {f: record.get(f) for f in fields}
return hash(frozenset(record_.items()))