mirror of
https://github.com/simple-login/app.git
synced 2024-11-16 00:48:32 +01:00
122 lines
3.1 KiB
Python
122 lines
3.1 KiB
Python
from typing import Optional, List, Tuple
|
|
|
|
import dns.resolver
|
|
|
|
|
|
def _get_dns_resolver():
|
|
my_resolver = dns.resolver.Resolver()
|
|
|
|
# 1.1.1.1 is CloudFlare's public DNS server
|
|
my_resolver.nameservers = ["1.1.1.1"]
|
|
|
|
return my_resolver
|
|
|
|
|
|
def get_ns(hostname) -> [str]:
|
|
try:
|
|
answers = _get_dns_resolver().resolve(hostname, "NS")
|
|
except Exception:
|
|
return []
|
|
return [a.to_text() for a in answers]
|
|
|
|
|
|
def get_cname_record(hostname) -> Optional[str]:
|
|
"""Return the CNAME record if exists for a domain, WITHOUT the trailing period at the end"""
|
|
try:
|
|
answers = _get_dns_resolver().query(hostname, "CNAME")
|
|
except Exception:
|
|
return None
|
|
|
|
for a in answers:
|
|
ret = a.to_text()
|
|
return ret[:-1]
|
|
|
|
return None
|
|
|
|
|
|
def get_mx_domains(hostname) -> [(int, str)]:
|
|
"""return list of (priority, domain name).
|
|
domain name ends with a "." at the end.
|
|
"""
|
|
try:
|
|
answers = _get_dns_resolver().query(hostname, "MX")
|
|
except Exception:
|
|
return []
|
|
|
|
ret = []
|
|
|
|
for a in answers:
|
|
record = a.to_text() # for ex '20 alt2.aspmx.l.google.com.'
|
|
parts = record.split(" ")
|
|
|
|
ret.append((int(parts[0]), parts[1]))
|
|
|
|
return ret
|
|
|
|
|
|
_include_spf = "include:"
|
|
|
|
|
|
def get_spf_domain(hostname) -> [str]:
|
|
"""return all domains listed in *include:*"""
|
|
try:
|
|
answers = _get_dns_resolver().query(hostname, "TXT")
|
|
except Exception:
|
|
return []
|
|
|
|
ret = []
|
|
|
|
for a in answers: # type: dns.rdtypes.ANY.TXT.TXT
|
|
for record in a.strings:
|
|
record = record.decode() # record is bytes
|
|
|
|
if record.startswith("v=spf1"):
|
|
parts = record.split(" ")
|
|
for part in parts:
|
|
if part.startswith(_include_spf):
|
|
ret.append(part[part.find(_include_spf) + len(_include_spf) :])
|
|
|
|
return ret
|
|
|
|
|
|
def get_txt_record(hostname) -> [str]:
|
|
"""return all domains listed in *include:*"""
|
|
try:
|
|
answers = _get_dns_resolver().query(hostname, "TXT")
|
|
except Exception:
|
|
return []
|
|
|
|
ret = []
|
|
|
|
for a in answers: # type: dns.rdtypes.ANY.TXT.TXT
|
|
for record in a.strings:
|
|
record = record.decode() # record is bytes
|
|
|
|
ret.append(record)
|
|
|
|
return ret
|
|
|
|
|
|
def is_mx_equivalent(
|
|
mx_domains: List[Tuple[int, str]], ref_mx_domains: List[Tuple[int, str]]
|
|
) -> bool:
|
|
"""
|
|
Compare mx_domains with ref_mx_domains to see if they are equivalent.
|
|
mx_domains and ref_mx_domains are list of (priority, domain)
|
|
|
|
The priority order is taken into account but not the priority number.
|
|
For example, [(1, domain1), (2, domain2)] is equivalent to [(10, domain1), (20, domain2)]
|
|
"""
|
|
mx_domains = sorted(mx_domains, key=lambda priority_domain: priority_domain[0])
|
|
ref_mx_domains = sorted(
|
|
ref_mx_domains, key=lambda priority_domain: priority_domain[0]
|
|
)
|
|
|
|
if len(mx_domains) != len(ref_mx_domains):
|
|
return False
|
|
|
|
for i in range(0, len(mx_domains)):
|
|
if mx_domains[i][1] != ref_mx_domains[i][1]:
|
|
return False
|
|
|
|
return True
|