|
| 1 | +import logging |
| 2 | +from enum import Enum |
| 3 | +from ipaddress import IPv4Address, IPv6Address |
| 4 | +from typing import Any, AsyncIterable, Dict, Iterable, List, NewType, Optional, Union |
| 5 | +from urllib.parse import urlparse |
| 6 | + |
| 7 | +import aiodns |
| 8 | +from pydantic import BaseModel, HttpUrl |
| 9 | + |
| 10 | +from .conf import settings |
| 11 | +from .exceptions import DomainConfigurationError |
| 12 | + |
| 13 | +logger = logging.getLogger(__name__) |
| 14 | + |
| 15 | +Hostname = NewType("Hostname", str) |
| 16 | + |
| 17 | + |
| 18 | +class TargetType(str, Enum): |
| 19 | + """ |
| 20 | + The type of target that a domain points to. |
| 21 | +
|
| 22 | + - IPFS: The domain points to an IPFS hash. |
| 23 | + - PROGRAM: The domain points to an aleph.im program. |
| 24 | + - INSTANCE: The domain points to an aleph.im instance. |
| 25 | + """ |
| 26 | + |
| 27 | + IPFS = "ipfs" |
| 28 | + PROGRAM = "program" |
| 29 | + INSTANCE = "instance" |
| 30 | + |
| 31 | + |
| 32 | +class DNSRule(BaseModel): |
| 33 | + """ |
| 34 | + A DNS rule is a DNS record that is required for a domain to be configured. |
| 35 | +
|
| 36 | + Args: |
| 37 | + name: The name of the rule. |
| 38 | + dns: The DNS record. |
| 39 | + info: Instructions to configure the DNS record. |
| 40 | + on_error: Error message when the rule is not found. |
| 41 | + """ |
| 42 | + |
| 43 | + name: str |
| 44 | + dns: Dict[str, Any] |
| 45 | + info: str |
| 46 | + on_error: str |
| 47 | + |
| 48 | + def raise_error(self, status: Dict[str, bool]): |
| 49 | + raise DomainConfigurationError((self.info, self.on_error, status)) |
| 50 | + |
| 51 | + |
| 52 | +def hostname_from_url(url: Union[HttpUrl, str]) -> Hostname: |
| 53 | + """Extract FQDN from url""" |
| 54 | + |
| 55 | + parsed = urlparse(url) |
| 56 | + if all([parsed.scheme, parsed.netloc]) is True: |
| 57 | + url = parsed.netloc |
| 58 | + |
| 59 | + return Hostname(url) |
| 60 | + |
| 61 | + |
| 62 | +async def get_target_type(fqdn: Hostname) -> Optional[TargetType]: |
| 63 | + """Returns the aleph.im target type of the domain""" |
| 64 | + domain_validator = DomainValidator() |
| 65 | + resolver = await domain_validator.get_resolver_for(fqdn) |
| 66 | + try: |
| 67 | + entry = await resolver.query(fqdn, "CNAME") |
| 68 | + cname = getattr(entry, "cname") |
| 69 | + if settings.DNS_IPFS_DOMAIN in cname: |
| 70 | + return TargetType.IPFS |
| 71 | + elif settings.DNS_PROGRAM_DOMAIN in cname: |
| 72 | + return TargetType.PROGRAM |
| 73 | + elif settings.DNS_INSTANCE_DOMAIN in cname: |
| 74 | + return TargetType.INSTANCE |
| 75 | + |
| 76 | + return None |
| 77 | + except aiodns.error.DNSError: |
| 78 | + return None |
| 79 | + |
| 80 | + |
| 81 | +class DomainValidator: |
| 82 | + """ |
| 83 | + Tools used to analyze domain names used on the aleph.im network. |
| 84 | + """ |
| 85 | + |
| 86 | + resolver: aiodns.DNSResolver |
| 87 | + |
| 88 | + def __init__(self): |
| 89 | + self.resolver = aiodns.DNSResolver(servers=settings.DNS_RESOLVERS) |
| 90 | + |
| 91 | + async def get_name_servers(self, hostname: Hostname): |
| 92 | + """Get DNS name servers (NS) of a domain""" |
| 93 | + dns_servers = settings.DNS_RESOLVERS |
| 94 | + fqdn = hostname |
| 95 | + |
| 96 | + while True: |
| 97 | + # Detect and get authoritative NS of subdomains if delegated |
| 98 | + try: |
| 99 | + entries = await self.resolver.query(fqdn, "NS") |
| 100 | + servers: List[Any] = [] |
| 101 | + for entry in entries: |
| 102 | + servers += await self.get_ipv6_addresses(entry.host) |
| 103 | + servers += await self.get_ipv4_addresses(entry.host) |
| 104 | + |
| 105 | + dns_servers = servers |
| 106 | + break |
| 107 | + except aiodns.error.DNSError: |
| 108 | + sub_domains = fqdn.split(".") |
| 109 | + if len(sub_domains) > 2: |
| 110 | + fqdn = Hostname(".".join(sub_domains[1:])) |
| 111 | + continue |
| 112 | + |
| 113 | + if len(sub_domains) == 2: |
| 114 | + break |
| 115 | + except Exception as err: |
| 116 | + logger.debug(f"Unexpected {err=}, {type(err)=}") |
| 117 | + break |
| 118 | + |
| 119 | + return dns_servers |
| 120 | + |
| 121 | + async def get_resolver_for(self, hostname: Hostname): |
| 122 | + dns_servers = await self.get_name_servers(hostname) |
| 123 | + return aiodns.DNSResolver(servers=dns_servers) |
| 124 | + |
| 125 | + async def get_ipv4_addresses(self, hostname: Hostname) -> List[IPv4Address]: |
| 126 | + """Returns all IPv4 addresses for a domain""" |
| 127 | + entries: Iterable = await self.resolver.query(hostname, "A") or [] |
| 128 | + return [entry.host for entry in entries] |
| 129 | + |
| 130 | + async def get_ipv6_addresses(self, hostname: Hostname) -> List[IPv6Address]: |
| 131 | + """Returns all IPv6 addresses for a domain""" |
| 132 | + entries: Iterable = await self.resolver.query(hostname, "AAAA") or [] |
| 133 | + return [entry.host for entry in entries] |
| 134 | + |
| 135 | + async def get_dnslinks(self, hostname: Hostname) -> List[str]: |
| 136 | + """Returns all DNSLink values for a domain.""" |
| 137 | + entries = await self.resolver.query(f"_dnslink.{hostname}", "TXT") |
| 138 | + return [entry.text for entry in entries] |
| 139 | + |
| 140 | + async def get_dnslink(self, hostname: Hostname) -> Optional[str]: |
| 141 | + """Returns the DNSLink corresponding to a domain. |
| 142 | +
|
| 143 | + Since it is possible to add multiple TXT records containing a DNSLink to |
| 144 | + the same domain, a behaviour has to be defined. |
| 145 | +
|
| 146 | + - Some IPFS implementations might use the first valid dnslink= record they find. |
| 147 | + - Others might throw an error indicating that the DNSLink resolution is ambiguous due to multiple records. |
| 148 | + - Still, others might try to fetch content from all provided DNSLinks, |
| 149 | + though this behavior would be less common and may introduce overhead. |
| 150 | + """ |
| 151 | + dnslinks = await self.get_dnslinks(hostname) |
| 152 | + return dnslinks[0] if dnslinks else None |
| 153 | + |
| 154 | + async def get_txt_values( |
| 155 | + self, hostname: Hostname, delimiter: Optional[str] = None |
| 156 | + ) -> AsyncIterable[str]: |
| 157 | + """Returns all TXT values for a domain""" |
| 158 | + entries: Iterable = await self.resolver.query(hostname, "TXT") or [] |
| 159 | + for entry in entries: |
| 160 | + if not hasattr(entry, "text"): |
| 161 | + logger.debug("An entry does not have any text") |
| 162 | + continue |
| 163 | + if not entry.text.startswith("0x"): |
| 164 | + logger.debug("Does not look like an Ethereum address") |
| 165 | + continue |
| 166 | + |
| 167 | + if delimiter: |
| 168 | + for part in entry.text.split(delimiter): |
| 169 | + yield part |
| 170 | + else: |
| 171 | + yield entry.text |
| 172 | + |
| 173 | + async def check_domain( |
| 174 | + self, hostname: Hostname, target: TargetType, owner: Optional[str] = None |
| 175 | + ) -> Dict: |
| 176 | + """ |
| 177 | + Checks that the domain points towards the given aleph.im target. |
| 178 | +
|
| 179 | + Args: |
| 180 | + hostname: The hostname of the domain. |
| 181 | + target: The aleph.im target type. |
| 182 | + owner: The owner wallet address of the domain for ownership proof. |
| 183 | +
|
| 184 | + Raises: |
| 185 | + DomainConfigurationError: If the domain is not configured. |
| 186 | +
|
| 187 | + Returns: |
| 188 | + A dictionary containing the status of the domain configuration. |
| 189 | + """ |
| 190 | + status = {"cname": False, "owner_proof": False} |
| 191 | + |
| 192 | + dns_rules = self.get_required_dns_rules(hostname, target, owner) |
| 193 | + |
| 194 | + for dns_rule in dns_rules: |
| 195 | + status[dns_rule.name] = False |
| 196 | + |
| 197 | + record_name = dns_rule.dns["name"] |
| 198 | + record_type = dns_rule.dns["type"] |
| 199 | + record_value = dns_rule.dns["value"] |
| 200 | + |
| 201 | + try: |
| 202 | + resolver = await self.get_resolver_for(hostname) |
| 203 | + entries = await resolver.query(record_name, record_type.upper()) |
| 204 | + except aiodns.error.DNSError: |
| 205 | + # Continue checks |
| 206 | + entries = None |
| 207 | + |
| 208 | + if entries and record_type == "txt": |
| 209 | + for entry in entries: |
| 210 | + if hasattr(entry, "text") and entry.text == record_value: |
| 211 | + break |
| 212 | + else: |
| 213 | + return dns_rule.raise_error(status) |
| 214 | + elif ( |
| 215 | + entries is None |
| 216 | + or not hasattr(entries, record_type) |
| 217 | + or getattr(entries, record_type) != record_value |
| 218 | + ): |
| 219 | + return dns_rule.raise_error(status) |
| 220 | + |
| 221 | + status[dns_rule.name] = True |
| 222 | + |
| 223 | + return status |
| 224 | + |
| 225 | + def get_required_dns_rules( |
| 226 | + self, hostname: Hostname, target: TargetType, owner: Optional[str] = None |
| 227 | + ) -> List[DNSRule]: |
| 228 | + """ |
| 229 | + Returns the DNS rules (CNAME, TXT) required for a domain to be configured. |
| 230 | +
|
| 231 | + Args: |
| 232 | + hostname: The hostname of the domain. |
| 233 | + target: The aleph.im target type. |
| 234 | + owner: The owner wallet address of the domain to add as an ownership proof. |
| 235 | +
|
| 236 | + Returns: |
| 237 | + A list of DNS rules with |
| 238 | + """ |
| 239 | + target = target.lower() |
| 240 | + dns_rules = [] |
| 241 | + |
| 242 | + cname_value = None |
| 243 | + if target == TargetType.IPFS: |
| 244 | + cname_value = settings.DNS_IPFS_DOMAIN |
| 245 | + elif target == TargetType.PROGRAM: |
| 246 | + cname_value = f"{hostname}.{settings.DNS_PROGRAM_DOMAIN}" |
| 247 | + elif target == TargetType.INSTANCE: |
| 248 | + cname_value = f"{hostname}.{settings.DNS_INSTANCE_DOMAIN}" |
| 249 | + |
| 250 | + # cname rule |
| 251 | + dns_rules.append( |
| 252 | + DNSRule( |
| 253 | + name="cname", |
| 254 | + dns={ |
| 255 | + "type": "cname", |
| 256 | + "name": hostname, |
| 257 | + "value": cname_value, |
| 258 | + }, |
| 259 | + info=f"Create a CNAME record for {hostname} with value {cname_value}", |
| 260 | + on_error=f"CNAME record not found: {hostname}", |
| 261 | + ) |
| 262 | + ) |
| 263 | + |
| 264 | + if target == TargetType.IPFS: |
| 265 | + # ipfs rule |
| 266 | + dns_rules.append( |
| 267 | + DNSRule( |
| 268 | + name="delegation", |
| 269 | + dns={ |
| 270 | + "type": "cname", |
| 271 | + "name": f"_dnslink.{hostname}", |
| 272 | + "value": f"_dnslink.{hostname}.{settings.DNS_STATIC_DOMAIN}", |
| 273 | + }, |
| 274 | + info=f"Create a CNAME record for _dnslink.{hostname} with value _dnslink.{hostname}.{settings.DNS_STATIC_DOMAIN}", |
| 275 | + on_error=f"CNAME record not found: _dnslink.{hostname}", |
| 276 | + ) |
| 277 | + ) |
| 278 | + |
| 279 | + if owner: |
| 280 | + # ownership rule |
| 281 | + dns_rules.append( |
| 282 | + DNSRule( |
| 283 | + name="owner_proof", |
| 284 | + dns={ |
| 285 | + "type": "txt", |
| 286 | + "name": f"_control.{hostname}", |
| 287 | + "value": owner, |
| 288 | + }, |
| 289 | + info=f"Create a TXT record for _control.{hostname} with value {owner}", |
| 290 | + on_error="Owner address mismatch", |
| 291 | + ) |
| 292 | + ) |
| 293 | + |
| 294 | + return dns_rules |
0 commit comments