dns_propagation

Instructions

Implementation Reference

• src/sounding/server.py:615-682 (handler)

  The dns_propagation tool handler. Queries multiple public DNS resolvers (Google, Cloudflare, OpenDNS, Quad9) plus system default in parallel, then checks consistency across results. Validates domain via sanitize_domain() and record_type against allowed set {A, AAAA, CNAME, MX, TXT}. Returns domain, record_type, consistent flag, and per-resolver results.

  @mcp.tool()
  async def dns_propagation(
      domain: str,
      record_type: str = "A",
  ) -> dict:
      """Check DNS propagation across multiple public resolvers.
  
      Queries Google (8.8.8.8), Cloudflare (1.1.1.1), OpenDNS (208.67.222.222),
      Quad9 (9.9.9.9), and the system default resolver in parallel, then
      highlights any inconsistencies between them.
  
      Supported record types: A, AAAA, CNAME, MX, TXT.
      """
      domain = sanitize_domain(domain)
      record_type = record_type.upper()
      allowed_types = {"A", "AAAA", "CNAME", "MX", "TXT"}
      if record_type not in allowed_types:
          raise ValueError(f"record_type must be one of {allowed_types}")
  
      async def _query(name: str, nameserver: str | None) -> dict:
          """Query a single resolver and return its results."""
          resolver = dns.resolver.Resolver()
          if nameserver:
              resolver.nameservers = [nameserver]
          resolver.lifetime = 10  # seconds
  
          try:
              answers = await asyncio.get_event_loop().run_in_executor(
                  None, lambda: resolver.resolve(domain, record_type)
              )
              records = sorted(str(r) for r in answers)
              ttl = answers.rrset.ttl if answers.rrset else None
              return {"resolver": name, "nameserver": nameserver or "system", "records": records, "ttl": ttl}
          except dns.resolver.NXDOMAIN:
              return {"resolver": name, "nameserver": nameserver or "system", "records": [], "error": "NXDOMAIN"}
          except dns.resolver.NoAnswer:
              return {"resolver": name, "nameserver": nameserver or "system", "records": [], "error": "No answer"}
          except dns.resolver.NoNameservers:
              return {"resolver": name, "nameserver": nameserver or "system", "records": [], "error": "No nameservers"}
          except Exception as exc:
              return {"resolver": name, "nameserver": nameserver or "system", "records": [], "error": str(exc)}
  
      # Build tasks: named resolvers + system default
      tasks = [_query(name, ns) for name, ns in _PUBLIC_RESOLVERS.items()]
      tasks.append(_query("System Default", None))
  
      results = await asyncio.gather(*tasks)
  
      # Detect inconsistencies — compare record sets across resolvers
      record_sets: dict[str, set[str]] = {}
      for r in results:
          if "error" not in r:
              record_sets[r["resolver"]] = set(r["records"])
  
      consistent = True
      if len(record_sets) > 1:
          reference = next(iter(record_sets.values()))
          for rs in record_sets.values():
              if rs != reference:
                  consistent = False
                  break
  
      return {
          "domain": domain,
          "record_type": record_type,
          "consistent": consistent,
          "resolvers": results,
      }

• src/sounding/server.py:615-615 (registration)

  The @mcp.tool() decorator registering dns_propagation as an MCP tool on the FastMCP 'sounding' server instance (defined on line 33).

  @mcp.tool()

• src/sounding/server.py:606-612 (schema)

  Public DNS resolver definitions and allowed record types used as input constraints for the dns_propagation tool.

  # Public DNS resolvers for propagation checks.
  _PUBLIC_RESOLVERS = {
      "Google": "8.8.8.8",
      "Cloudflare": "1.1.1.1",
      "OpenDNS": "208.67.222.222",
      "Quad9": "9.9.9.9",
  }

• src/sounding/validators.py:219-235 (helper)

  The sanitize_domain helper used by dns_propagation to clean and validate the domain input, stripping whitespace, lowercasing, and checking against shell metacharacters and hostname patterns.

  def sanitize_domain(domain: str) -> str:
      """Sanitize a domain name for DNS lookups.
  
      Strips whitespace, lowercases, and rejects shell metacharacters.
      Returns the cleaned domain.
      """
      domain = domain.strip().lower()
      if not domain:
          raise ValueError("Domain must not be empty")
  
      if _SHELL_META.search(domain):
          raise ValueError(f"Domain contains forbidden characters: {domain!r}")
  
      if not _HOSTNAME_RE.match(domain):
          raise ValueError(f"Invalid domain: {domain!r}")
  
      return domain