1"""Domain registration (RDAP, WHOIS fallback), DNS records (DNS over HTTPS) and TLS certificate checks."""
2from __future__ import annotations
3
4import asyncio
5import datetime as dt
6import re
7import socket
8import ssl
9
10import httpx
11import tldextract
12from cryptography import x509
13from cryptography.x509.oid import ExtensionOID, NameOID
14
15RDAP_BOOTSTRAP = 'https://data.iana.org/rdap/dns.json'
16DOH = 'https://cloudflare-dns.com/dns-query'
17DNS_TYPES = ('A', 'AAAA', 'MX', 'NS', 'TXT', 'CNAME', 'CAA', 'SOA')
18WHOIS_FIELDS = {
19 'registrar': r'(?:Registrar|registrar|Sponsoring Registrar|registrar name)\s*:\s*(.+)',
20 'createdDate': r'(?:Creation Date|created|Registered on|Registration Time|Created On|domain_dateregistered)\s*:\s*(.+)',
21 'updatedDate': r'(?:Updated Date|changed|Last updated|modified|Last Modified)\s*:\s*(.+)',
22 'expiresDate': r'(?:Registry Expiry Date|Expiration Date|Expiry date|expires|paid-till|Expiration Time|Expiry Date)\s*:\s*(.+)',
23}
24NOT_FOUND = re.compile(r'no match|not found|no data found|no entries found|status:\s*free|is available|'
25 r'domain not found|object does not exist|no object found', re.I)
26
27
28_PSL = tldextract.TLDExtract(suffix_list_urls=(), cache_dir=None)
29
30
31class DomainError(Exception):
32 pass
33
34
35def registrable(domain: str) -> str:
36 """expired.badssl.com -> badssl.com, www.bbc.co.uk -> bbc.co.uk: registries know only the registered name."""
37 return _PSL(domain).top_domain_under_public_suffix or domain
38
39
40def normalize(raw: str) -> str:
41 d = raw.strip().lower()
42 d = re.sub(r'^[a-z][a-z0-9+.-]*://', '', d).split('/')[0].split('?')[0].split(':')[0].strip('.')
43 if d.startswith('www.'):
44 d = d[4:]
45 try:
46 d = d.encode('idna').decode('ascii')
47 except UnicodeError as e:
48 raise DomainError(f'"{raw}" is not a valid domain') from e
49 if not re.fullmatch(r'(?=.{1,253}$)([a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?\.)+[a-z0-9-]{2,63}', d):
50 raise DomainError(f'"{raw}" is not a valid domain')
51 return d
52
53
54class Rdap:
55 def __init__(self, client: httpx.AsyncClient):
56 self.client = client
57 self.servers: dict[str, str] | None = None
58 self.whois_servers: dict[str, str | None] = {}
59
60 async def _bootstrap(self) -> dict[str, str]:
61 if self.servers is None:
62 r = await self.client.get(RDAP_BOOTSTRAP)
63 r.raise_for_status()
64 self.servers = {tld: urls[0] for tlds, urls in r.json()['services'] for tld in tlds}
65 return self.servers
66
67 async def lookup(self, domain: str) -> dict:
68 """Registration data; falls back to WHOIS for TLDs without RDAP (e.g. .io, .de)."""
69 tld = domain.rsplit('.', 1)[-1]
70 server = (await self._bootstrap()).get(tld)
71 if server:
72 try:
73 r = await self.client.get(server.rstrip('/') + '/domain/' + domain,
74 headers={'Accept': 'application/rdap+json'}, follow_redirects=True)
75 if r.status_code == 404:
76 return {'registered': False, 'source': 'rdap'}
77 if r.status_code < 400:
78 return {'registered': True, 'source': 'rdap', **parse_rdap(r.json())}
79 except (httpx.HTTPError, ValueError):
80 pass
81 return await self.whois(domain)
82
83 async def whois(self, domain: str) -> dict:
84 tld = domain.rsplit('.', 1)[-1]
85 if tld not in self.whois_servers:
86 text = await asyncio.to_thread(whois_query, 'whois.iana.org', tld)
87 m = re.search(r'^(?:refer|whois):\s*(\S+)', text, re.M)
88 self.whois_servers[tld] = m.group(1) if m else None
89 server = self.whois_servers[tld]
90 if not server:
91 raise DomainError(f'no RDAP or WHOIS service is known for .{tld}')
92 text = await asyncio.to_thread(whois_query, server, domain)
93 return {'source': 'whois', **parse_whois(text)}
94
95
96def _connect(host: str, port: int, timeout: float) -> socket.socket:
97 """IPv4 addresses first: Apify has no IPv6 route, and some servers list IPv6 first."""
98 try:
99 infos = socket.getaddrinfo(host, port, type=socket.SOCK_STREAM)
100 except socket.gaierror as e:
101 raise OSError(f'cannot resolve {host}') from e
102 infos.sort(key=lambda i: i[0] != socket.AF_INET)
103 last = None
104 for family, stype, proto, _, addr in infos:
105 sock = socket.socket(family, stype, proto)
106 sock.settimeout(timeout)
107 try:
108 sock.connect(addr)
109 return sock
110 except OSError as e:
111 sock.close()
112 last = e
113 raise last or OSError(f'no address for {host}')
114
115
116def whois_query(server: str, query: str, timeout: float = 15) -> str:
117 for attempt in range(2):
118 try:
119 with _connect(server, 43, timeout) as s:
120 s.sendall(query.encode() + b'\r\n')
121 chunks = []
122 while chunk := s.recv(65536):
123 chunks.append(chunk)
124 if sum(map(len, chunks)) > 1 << 20:
125 break
126 return b''.join(chunks).decode('utf-8', 'replace')
127 except OSError as e:
128 if attempt:
129 raise DomainError(f'WHOIS server {server} did not answer: {e}') from e
130 return ''
131
132
133
134def _date(s: str | None) -> str | None:
135 if not s:
136 return None
137 s = s.strip()
138 for fmt in ('%Y-%m-%dT%H:%M:%SZ', '%Y-%m-%dT%H:%M:%S.%fZ', '%Y-%m-%dT%H:%M:%S%z', '%Y-%m-%d %H:%M:%S',
139 '%Y-%m-%d', '%d-%b-%Y', '%Y.%m.%d', '%d.%m.%Y', '%Y/%m/%d', '%Y-%m-%dT%H:%M:%S'):
140 try:
141 d = dt.datetime.strptime(s.split(' (')[0][:32], fmt)
142 return d.date().isoformat()
143 except ValueError:
144 continue
145 m = re.search(r'\d{4}-\d{2}-\d{2}', s)
146 return m.group(0) if m else s[:40]
147
148
149def _days_left(date: str | None) -> int | None:
150 try:
151 return (dt.date.fromisoformat(date) - dt.date.today()).days if date else None
152 except ValueError:
153 return None
154
155
156def parse_rdap(d: dict) -> dict:
157 events = {e.get('eventAction'): e.get('eventDate') for e in d.get('events', [])}
158 registrar, abuse = None, None
159 org = country = None
160 for ent in d.get('entities', []):
161 roles = ent.get('roles', [])
162 vcard = {f[0]: f[3] for f in (ent.get('vcardArray') or [None, []])[1] if len(f) > 3}
163 if 'registrar' in roles:
164 registrar = vcard.get('fn') or ent.get('handle')
165 for sub in ent.get('entities', []):
166 if 'abuse' in sub.get('roles', []):
167 sv = {f[0]: f[3] for f in (sub.get('vcardArray') or [None, []])[1] if len(f) > 3}
168 abuse = sv.get('email')
169 if 'registrant' in roles:
170 org = vcard.get('org') or vcard.get('fn')
171 adr = vcard.get('adr')
172 if isinstance(adr, list) and adr:
173 country = adr[-1] or None
174 expires = _date(events.get('expiration'))
175 return {
176 'registrar': registrar, 'registrarAbuseEmail': abuse,
177 'createdDate': _date(events.get('registration')), 'updatedDate': _date(events.get('last changed')),
178 'expiresDate': expires, 'daysUntilExpiry': _days_left(expires),
179 'status': d.get('status', []),
180 'nameservers': sorted(n.get('ldhName', '').lower() for n in d.get('nameservers', []) if n.get('ldhName')),
181 'dnssec': (d.get('secureDNS') or {}).get('delegationSigned'),
182 'registrantOrganization': org if org and 'redacted' not in str(org).lower() else None,
183 'registrantCountry': country if isinstance(country, str) and len(country) <= 40 else None,
184 }
185
186
187def parse_whois(text: str) -> dict:
188 if NOT_FOUND.search(text[:2000]) and not re.search(r'Creation Date|created:', text, re.I):
189 return {'registered': False}
190 out = {'registered': True}
191 for key, pattern in WHOIS_FIELDS.items():
192 m = re.search(pattern, text, re.I)
193 out[key] = (_date(m.group(1)) if key.endswith('Date') else m.group(1).strip()[:200]) if m else None
194 out['daysUntilExpiry'] = _days_left(out.get('expiresDate'))
195 ns = re.findall(r'(?:Name Server|nserver|Nameservers?)\s*:\s*([a-z0-9.-]+)', text, re.I)
196 out['nameservers'] = sorted({n.lower().rstrip('.') for n in ns})
197 out['status'] = sorted(set(re.findall(r'(?:Domain Status|status)\s*:\s*([A-Za-z]+)', text)))[:10]
198 return out
199
200
201async def dns_records(client: httpx.AsyncClient, domain: str, types=DNS_TYPES) -> dict:
202 async def one(name, rtype):
203 r = await client.get(DOH, params={'name': name, 'type': rtype}, headers={'Accept': 'application/dns-json'})
204 r.raise_for_status()
205 data = r.json()
206 code = {'A': 1, 'NS': 2, 'CNAME': 5, 'SOA': 6, 'MX': 15, 'TXT': 16, 'AAAA': 28, 'CAA': 257}[rtype]
207 out = []
208 for a in data.get('Answer', []):
209 if a.get('type') != code:
210 continue
211 v = re.sub(r'"\s*"', '', a['data']).strip('"') if rtype == 'TXT' else a['data']
212 out.append(v.rstrip('.') if rtype in ('MX', 'NS', 'CNAME') else v)
213 return out
214 results = await asyncio.gather(*(one(domain, t) for t in types), one('_dmarc.' + domain, 'TXT'),
215 return_exceptions=True)
216 out = {}
217 for t, res in zip((*types, 'DMARC'), results):
218 out[t] = [] if isinstance(res, Exception) else res
219 out['SPF'] = next((t for t in out.get('TXT', []) if t.lower().startswith('v=spf1')), None)
220 out['DMARC'] = next((t for t in out.pop('DMARC') if t.lower().startswith('v=dmarc1')), None)
221 out['MX'] = sorted(out.get('MX', []), key=lambda m: int(m.split()[0]) if m.split()[0].isdigit() else 99)
222 return out
223
224
225def tls_certificate(host: str, port: int = 443, timeout: float = 10) -> dict:
226 """Certificate of host:443, checked against the system trust store; details are read even when invalid.
227 A connection reset is retried once (some servers drop the first handshake)."""
228 res = _tls_certificate(host, port, timeout)
229 if not res.get('reachable') and 'reset' in res.get('error', '').lower():
230 res = _tls_certificate(host, port, timeout)
231 return res
232
233
234def _tls_certificate(host: str, port: int, timeout: float) -> dict:
235 verify_error = None
236 der = None
237 ctx = ssl.create_default_context()
238 try:
239 with _connect(host, port, timeout) as sock:
240 with ctx.wrap_socket(sock, server_hostname=host) as tls:
241 der, proto = tls.getpeercert(binary_form=True), tls.version()
242 except ssl.SSLCertVerificationError as e:
243 verify_error = e.verify_message or str(e)
244 except (OSError, ssl.SSLError) as e:
245 return {'reachable': False, 'error': str(e)[:200]}
246 if der is None:
247 insecure = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
248 insecure.check_hostname = False
249 insecure.verify_mode = ssl.CERT_NONE
250 try:
251 with _connect(host, port, timeout) as sock:
252 with insecure.wrap_socket(sock, server_hostname=host) as tls:
253 der, proto = tls.getpeercert(binary_form=True), tls.version()
254 except (OSError, ssl.SSLError) as e:
255 return {'reachable': False, 'error': str(e)[:200]}
256 cert = x509.load_der_x509_certificate(der)
257
258 def name(n, oid):
259 v = n.get_attributes_for_oid(oid)
260 return v[0].value if v else None
261 try:
262 sans = cert.extensions.get_extension_for_oid(ExtensionOID.SUBJECT_ALTERNATIVE_NAME).value.get_values_for_type(
263 x509.DNSName)
264 except x509.ExtensionNotFound:
265 sans = []
266 not_after = cert.not_valid_after_utc
267 return {
268 'reachable': True, 'valid': verify_error is None, 'verifyError': verify_error, 'protocol': proto,
269 'subject': name(cert.subject, NameOID.COMMON_NAME),
270 'issuer': name(cert.issuer, NameOID.ORGANIZATION_NAME) or name(cert.issuer, NameOID.COMMON_NAME),
271 'validFrom': cert.not_valid_before_utc.date().isoformat(), 'validTo': not_after.date().isoformat(),
272 'daysLeft': (not_after - dt.datetime.now(dt.timezone.utc)).days,
273 'san': sans[:100],
274 }
