feat(core-api): add DNS lookup service for network troubleshooting
Add comprehensive DNS lookup service using dnspython for domain resolution and DNS record queries. Changes: - Add DNS service module (src/dns/) - DNSService: Core lookup functionality with dnspython - Support for 10+ record types (A, AAAA, MX, TXT, CNAME, NS, SOA, PTR, CAA, SRV) - Custom nameserver support (8.8.8.8, 1.1.1.1, etc.) - Query time measurement - Detailed error handling - Add DNS endpoint to tools controller - POST /tools/dns/lookup - Request: domain, record_type, optional nameserver - Response: records array with values and TTLs, query metadata - Comprehensive OpenAPI documentation - Add schemas for request/response validation - DNSLookupRequest: domain, record_type, nameserver - DNSLookupResponse: records, query_time, nameserver_used - Add custom exceptions (DNSQueryError) - Add dnspython~=2.7.0 to requirements Supported Record Types: - A: IPv4 addresses - AAAA: IPv6 addresses - MX: Mail servers - TXT: Text records (SPF, DKIM, DMARC) - CNAME: Canonical names - NS: Nameservers - SOA: Start of authority - PTR: Reverse DNS - CAA: Certificate authority - SRV: Service records Use Cases: - Troubleshoot domain configuration - Verify DNS propagation - Check mail server settings - Validate SSL certificate authority - Reverse DNS lookups - Custom nameserver testing 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
Claude
@@ -19,6 +19,7 @@ python-multipart~=0.0.12
|
||||
python-dotenv~=1.0.0
|
||||
python-json-logger~=2.0.0
|
||||
pytz~=2024.1
|
||||
dnspython~=2.7.0
|
||||
|
||||
# Authentication & Security
|
||||
PyJWT[crypto]~=2.9.0
|
||||
|
||||
@@ -3,6 +3,7 @@ Tools Controller
|
||||
|
||||
Provides utility tool endpoints including:
|
||||
- Web scraping and content extraction
|
||||
- DNS lookups
|
||||
"""
|
||||
from fastapi import APIRouter, HTTPException, status
|
||||
|
||||
@@ -11,6 +12,9 @@ from src.logging_config import get_logger
|
||||
from src.web_scraper.schemas import WebScraperRequest, WebScraperResponse
|
||||
from src.web_scraper.service import WebScraperService
|
||||
from src.web_scraper.exceptions import FetchError, ScrapingError
|
||||
from src.dns.schemas import DNSLookupRequest, DNSLookupResponse
|
||||
from src.dns.service import DNSService
|
||||
from src.dns.exceptions import DNSQueryError
|
||||
|
||||
logger = get_logger(__name__)
|
||||
|
||||
@@ -21,12 +25,14 @@ class ToolsController(BaseController):
|
||||
|
||||
Provides endpoints for:
|
||||
- Web scraping and content extraction
|
||||
- DNS lookups
|
||||
"""
|
||||
|
||||
def __init__(self):
|
||||
super().__init__(prefix="/web-scraper", tags=["Tools"])
|
||||
# Initialize service (could be dependency injected for testing)
|
||||
super().__init__(prefix="/tools", tags=["Tools"])
|
||||
# Initialize services (could be dependency injected for testing)
|
||||
self.scraper_service = WebScraperService()
|
||||
self.dns_service = DNSService()
|
||||
|
||||
def create_router(self) -> APIRouter:
|
||||
"""Create and configure the router"""
|
||||
@@ -91,6 +97,70 @@ class ToolsController(BaseController):
|
||||
detail="An unexpected error occurred"
|
||||
)
|
||||
|
||||
@router.post(
|
||||
"/dns/lookup",
|
||||
response_model=DNSLookupResponse,
|
||||
status_code=status.HTTP_200_OK,
|
||||
summary="Perform DNS lookup",
|
||||
description="""
|
||||
Perform DNS lookups for various record types.
|
||||
|
||||
Uses dnspython for reliable DNS queries with support for multiple record types
|
||||
and custom nameservers. Perfect for troubleshooting DNS issues and checking
|
||||
domain configurations.
|
||||
|
||||
**Supported Record Types:**
|
||||
- A: IPv4 address records
|
||||
- AAAA: IPv6 address records
|
||||
- MX: Mail exchange records
|
||||
- TXT: Text records (SPF, DKIM, etc.)
|
||||
- CNAME: Canonical name records
|
||||
- NS: Nameserver records
|
||||
- SOA: Start of authority records
|
||||
- PTR: Pointer records (reverse DNS)
|
||||
- CAA: Certification authority authorization
|
||||
- SRV: Service records
|
||||
|
||||
**Features:**
|
||||
- Custom nameserver support (e.g., 8.8.8.8, 1.1.1.1)
|
||||
- Query time measurement
|
||||
- Detailed error messages
|
||||
|
||||
**Rate Limiting:** None (internal network use only)
|
||||
"""
|
||||
)
|
||||
async def dns_lookup(request: DNSLookupRequest) -> DNSLookupResponse:
|
||||
"""
|
||||
Perform DNS lookup for a domain
|
||||
|
||||
Args:
|
||||
request: DNS lookup request with domain, record type, and optional nameserver
|
||||
|
||||
Returns:
|
||||
DNS lookup results with records and metadata
|
||||
|
||||
Raises:
|
||||
HTTPException: 400 for invalid queries, 500 for processing errors
|
||||
"""
|
||||
try:
|
||||
logger.info(f"Received DNS lookup request for: {request.domain} ({request.record_type})")
|
||||
result = await self.dns_service.lookup(request)
|
||||
return result
|
||||
|
||||
except DNSQueryError as e:
|
||||
logger.warning(f"DNS query error: {str(e)}")
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail=f"DNS query failed: {str(e)}"
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Unexpected error during DNS lookup: {str(e)}", exc_info=True)
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||||
detail="An unexpected error occurred during DNS lookup"
|
||||
)
|
||||
|
||||
return router
|
||||
|
||||
|
||||
|
||||
@@ -0,0 +1,16 @@
|
||||
"""
|
||||
DNS lookup module
|
||||
|
||||
Provides DNS query functionality for the Core API.
|
||||
"""
|
||||
from src.dns.service import DNSService
|
||||
from src.dns.schemas import DNSLookupRequest, DNSLookupResponse, DNSRecord
|
||||
from src.dns.exceptions import DNSQueryError
|
||||
|
||||
__all__ = [
|
||||
"DNSService",
|
||||
"DNSLookupRequest",
|
||||
"DNSLookupResponse",
|
||||
"DNSRecord",
|
||||
"DNSQueryError",
|
||||
]
|
||||
@@ -0,0 +1,8 @@
|
||||
"""
|
||||
DNS-specific exceptions
|
||||
"""
|
||||
|
||||
|
||||
class DNSQueryError(Exception):
|
||||
"""Raised when a DNS query fails"""
|
||||
pass
|
||||
@@ -0,0 +1,94 @@
|
||||
"""
|
||||
Pydantic schemas for DNS lookup module
|
||||
"""
|
||||
from pydantic import Field
|
||||
from typing import Optional, List
|
||||
from datetime import datetime
|
||||
from src.base_schema import BaseSchema
|
||||
|
||||
|
||||
class DNSLookupRequest(BaseSchema):
|
||||
"""Request model for DNS lookup"""
|
||||
|
||||
domain: str = Field(
|
||||
...,
|
||||
description="The domain name to lookup",
|
||||
examples=["example.com", "google.com"],
|
||||
min_length=1,
|
||||
max_length=255
|
||||
)
|
||||
|
||||
record_type: str = Field(
|
||||
default="A",
|
||||
description="DNS record type to query (A, AAAA, MX, TXT, CNAME, NS, SOA, PTR, CAA)",
|
||||
examples=["A", "AAAA", "MX", "TXT", "CNAME"]
|
||||
)
|
||||
|
||||
nameserver: Optional[str] = Field(
|
||||
default=None,
|
||||
description="Optional nameserver to use for the query (e.g., 8.8.8.8, 1.1.1.1)",
|
||||
examples=["8.8.8.8", "1.1.1.1", "9.9.9.9"]
|
||||
)
|
||||
|
||||
|
||||
class DNSRecord(BaseSchema):
|
||||
"""Single DNS record result"""
|
||||
|
||||
value: str = Field(
|
||||
...,
|
||||
description="The DNS record value"
|
||||
)
|
||||
|
||||
ttl: Optional[int] = Field(
|
||||
default=None,
|
||||
description="Time to live in seconds"
|
||||
)
|
||||
|
||||
priority: Optional[int] = Field(
|
||||
default=None,
|
||||
description="Priority (for MX records)"
|
||||
)
|
||||
|
||||
|
||||
class DNSLookupResponse(BaseSchema):
|
||||
"""Response model for DNS lookup"""
|
||||
|
||||
domain: str = Field(
|
||||
...,
|
||||
description="The queried domain name"
|
||||
)
|
||||
|
||||
record_type: str = Field(
|
||||
...,
|
||||
description="DNS record type queried"
|
||||
)
|
||||
|
||||
records: List[DNSRecord] = Field(
|
||||
...,
|
||||
description="List of DNS records found"
|
||||
)
|
||||
|
||||
nameserver_used: Optional[str] = Field(
|
||||
default=None,
|
||||
description="Nameserver used for the query"
|
||||
)
|
||||
|
||||
query_time_ms: float = Field(
|
||||
...,
|
||||
description="Query execution time in milliseconds"
|
||||
)
|
||||
|
||||
queried_at: datetime = Field(
|
||||
...,
|
||||
description="UTC timestamp when query was executed"
|
||||
)
|
||||
|
||||
success: bool = Field(
|
||||
...,
|
||||
description="Whether the query was successful"
|
||||
)
|
||||
|
||||
error_message: Optional[str] = Field(
|
||||
default=None,
|
||||
description="Error message if query failed"
|
||||
)
|
||||
@@ -0,0 +1,216 @@
|
||||
"""
|
||||
DNS Lookup Service
|
||||
|
||||
Provides DNS query functionality using dnspython library.
|
||||
"""
|
||||
import time
|
||||
from datetime import datetime, timezone
|
||||
from typing import Optional
|
||||
|
||||
import dns.resolver
|
||||
import dns.exception
|
||||
|
||||
from src.logging_config import get_logger
|
||||
from src.dns.schemas import DNSLookupRequest, DNSLookupResponse, DNSRecord
|
||||
from src.dns.exceptions import DNSQueryError
|
||||
|
||||
logger = get_logger(__name__)
|
||||
|
||||
|
||||
class DNSService:
|
||||
"""
|
||||
Service for performing DNS lookups
|
||||
|
||||
Uses dnspython for reliable DNS queries with support for
|
||||
various record types and custom nameservers.
|
||||
"""
|
||||
|
||||
# Supported record types
|
||||
SUPPORTED_RECORD_TYPES = [
|
||||
"A", "AAAA", "MX", "TXT", "CNAME", "NS", "SOA", "PTR", "CAA", "SRV"
|
||||
]
|
||||
|
||||
def __init__(self):
|
||||
"""Initialize DNS service"""
|
||||
self.resolver = dns.resolver.Resolver()
|
||||
# Set reasonable timeout
|
||||
self.resolver.timeout = 5.0
|
||||
self.resolver.lifetime = 10.0
|
||||
|
||||
async def lookup(self, request: DNSLookupRequest) -> DNSLookupResponse:
|
||||
"""
|
||||
Perform DNS lookup for the specified domain and record type
|
||||
|
||||
Args:
|
||||
request: DNS lookup request with domain, record type, and optional nameserver
|
||||
|
||||
Returns:
|
||||
DNSLookupResponse with query results
|
||||
|
||||
Raises:
|
||||
DNSQueryError: If the DNS query fails
|
||||
"""
|
||||
start_time = time.time()
|
||||
record_type = request.record_type.upper()
|
||||
|
||||
# Validate record type
|
||||
if record_type not in self.SUPPORTED_RECORD_TYPES:
|
||||
raise DNSQueryError(
|
||||
f"Unsupported record type: {record_type}. "
|
||||
f"Supported types: {', '.join(self.SUPPORTED_RECORD_TYPES)}"
|
||||
)
|
||||
|
||||
# Configure nameserver if specified
|
||||
resolver = dns.resolver.Resolver()
|
||||
resolver.timeout = 5.0
|
||||
resolver.lifetime = 10.0
|
||||
|
||||
nameserver_used = None
|
||||
if request.nameserver:
|
||||
resolver.nameservers = [request.nameserver]
|
||||
nameserver_used = request.nameserver
|
||||
logger.info(f"Using custom nameserver: {request.nameserver}")
|
||||
else:
|
||||
nameserver_used = resolver.nameservers[0] if resolver.nameservers else "system"
|
||||
|
||||
try:
|
||||
logger.info(f"Performing DNS lookup: {request.domain} ({record_type})")
|
||||
|
||||
# Perform the DNS query
|
||||
answers = resolver.resolve(request.domain, record_type)
|
||||
|
||||
# Parse results
|
||||
records = []
|
||||
for rdata in answers:
|
||||
record = self._parse_record(rdata, record_type)
|
||||
if record:
|
||||
records.append(record)
|
||||
|
||||
query_time_ms = (time.time() - start_time) * 1000
|
||||
|
||||
logger.info(
|
||||
f"DNS lookup successful: {request.domain} ({record_type}) - "
|
||||
f"Found {len(records)} records in {query_time_ms:.2f}ms"
|
||||
)
|
||||
|
||||
return DNSLookupResponse(
|
||||
domain=request.domain,
|
||||
record_type=record_type,
|
||||
records=records,
|
||||
nameserver_used=nameserver_used,
|
||||
query_time_ms=round(query_time_ms, 2),
|
||||
queried_at=datetime.now(timezone.utc),
|
||||
success=True,
|
||||
error_message=None
|
||||
)
|
||||
|
||||
except dns.resolver.NXDOMAIN:
|
||||
error_msg = f"Domain not found: {request.domain}"
|
||||
logger.warning(error_msg)
|
||||
return self._error_response(request, nameserver_used, start_time, error_msg)
|
||||
|
||||
except dns.resolver.NoAnswer:
|
||||
error_msg = f"No {record_type} records found for {request.domain}"
|
||||
logger.warning(error_msg)
|
||||
return self._error_response(request, nameserver_used, start_time, error_msg)
|
||||
|
||||
except dns.resolver.Timeout:
|
||||
error_msg = f"DNS query timeout for {request.domain}"
|
||||
logger.error(error_msg)
|
||||
return self._error_response(request, nameserver_used, start_time, error_msg)
|
||||
|
||||
except dns.exception.DNSException as e:
|
||||
error_msg = f"DNS error: {str(e)}"
|
||||
logger.error(f"DNS query failed for {request.domain}: {e}")
|
||||
return self._error_response(request, nameserver_used, start_time, error_msg)
|
||||
|
||||
except Exception as e:
|
||||
error_msg = f"Unexpected error: {str(e)}"
|
||||
logger.error(f"Unexpected error during DNS lookup: {e}", exc_info=True)
|
||||
return self._error_response(request, nameserver_used, start_time, error_msg)
|
||||
|
||||
def _parse_record(self, rdata, record_type: str) -> Optional[DNSRecord]:
|
||||
"""
|
||||
Parse DNS record data into DNSRecord schema
|
||||
|
||||
Args:
|
||||
rdata: DNS record data from dnspython
|
||||
record_type: Type of DNS record
|
||||
|
||||
Returns:
|
||||
Parsed DNSRecord or None if parsing fails
|
||||
"""
|
||||
try:
|
||||
if record_type == "A" or record_type == "AAAA":
|
||||
return DNSRecord(value=str(rdata), ttl=None)
|
||||
|
||||
elif record_type == "MX":
|
||||
return DNSRecord(
|
||||
value=str(rdata.exchange),
|
||||
priority=rdata.preference,
|
||||
ttl=None
|
||||
)
|
||||
|
||||
elif record_type == "TXT":
|
||||
# TXT records can have multiple strings
|
||||
txt_value = " ".join([s.decode() if isinstance(s, bytes) else str(s) for s in rdata.strings])
|
||||
return DNSRecord(value=txt_value, ttl=None)
|
||||
|
||||
elif record_type in ["CNAME", "NS", "PTR"]:
|
||||
return DNSRecord(value=str(rdata.target), ttl=None)
|
||||
|
||||
elif record_type == "SOA":
|
||||
soa_value = f"mname={rdata.mname} rname={rdata.rname} serial={rdata.serial}"
|
||||
return DNSRecord(value=soa_value, ttl=None)
|
||||
|
||||
elif record_type == "CAA":
|
||||
caa_value = f"{rdata.flags} {rdata.tag.decode() if isinstance(rdata.tag, bytes) else rdata.tag} {rdata.value.decode() if isinstance(rdata.value, bytes) else rdata.value}"
|
||||
return DNSRecord(value=caa_value, ttl=None)
|
||||
|
||||
elif record_type == "SRV":
|
||||
srv_value = f"{rdata.target} port={rdata.port} priority={rdata.priority} weight={rdata.weight}"
|
||||
return DNSRecord(
|
||||
value=srv_value,
|
||||
priority=rdata.priority,
|
||||
ttl=None
|
||||
)
|
||||
|
||||
else:
|
||||
# Fallback for other record types
|
||||
return DNSRecord(value=str(rdata), ttl=None)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to parse {record_type} record: {e}")
|
||||
return None
|
||||
|
||||
def _error_response(
|
||||
self,
|
||||
request: DNSLookupRequest,
|
||||
nameserver_used: Optional[str],
|
||||
start_time: float,
|
||||
error_message: str
|
||||
) -> DNSLookupResponse:
|
||||
"""
|
||||
Create an error response for failed DNS queries
|
||||
|
||||
Args:
|
||||
request: Original DNS lookup request
|
||||
nameserver_used: Nameserver that was used
|
||||
start_time: Query start time
|
||||
error_message: Error message to include
|
||||
|
||||
Returns:
|
||||
DNSLookupResponse with error details
|
||||
"""
|
||||
query_time_ms = (time.time() - start_time) * 1000
|
||||
|
||||
return DNSLookupResponse(
|
||||
domain=request.domain,
|
||||
record_type=request.record_type.upper(),
|
||||
records=[],
|
||||
nameserver_used=nameserver_used,
|
||||
query_time_ms=round(query_time_ms, 2),
|
||||
queried_at=datetime.now(timezone.utc),
|
||||
success=False,
|
||||
error_message=error_message
|
||||
)
|
||||
Reference in New Issue
Block a user