577 lines
23 KiB
Python
577 lines
23 KiB
Python
# dnsrecon/providers/crtsh_provider.py
|
|
|
|
import json
|
|
import re
|
|
from pathlib import Path
|
|
from typing import List, Dict, Any, Set
|
|
from urllib.parse import quote
|
|
from datetime import datetime, timezone
|
|
import requests
|
|
|
|
from .base_provider import BaseProvider
|
|
from core.provider_result import ProviderResult
|
|
from utils.helpers import _is_valid_domain
|
|
|
|
|
|
class CrtShProvider(BaseProvider):
|
|
"""
|
|
Provider for querying crt.sh certificate transparency database.
|
|
Now returns standardized ProviderResult objects with caching support.
|
|
"""
|
|
|
|
def __init__(self, name=None, session_config=None):
|
|
"""Initialize CrtSh provider with session-specific configuration."""
|
|
super().__init__(
|
|
name="crtsh",
|
|
rate_limit=60,
|
|
timeout=15,
|
|
session_config=session_config
|
|
)
|
|
self.base_url = "https://crt.sh/"
|
|
self._stop_event = None
|
|
|
|
# Initialize cache directory
|
|
self.cache_dir = Path('cache') / 'crtsh'
|
|
self.cache_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
# Compile regex for date filtering for efficiency
|
|
self.date_pattern = re.compile(r'^\d{4}-\d{2}-\d{2}[ T]\d{2}:\d{2}:\d{2}')
|
|
|
|
def get_name(self) -> str:
|
|
"""Return the provider name."""
|
|
return "crtsh"
|
|
|
|
def get_display_name(self) -> str:
|
|
"""Return the provider display name for the UI."""
|
|
return "crt.sh"
|
|
|
|
def requires_api_key(self) -> bool:
|
|
"""Return True if the provider requires an API key."""
|
|
return False
|
|
|
|
def get_eligibility(self) -> Dict[str, bool]:
|
|
"""Return a dictionary indicating if the provider can query domains and/or IPs."""
|
|
return {'domains': True, 'ips': False}
|
|
|
|
def is_available(self) -> bool:
|
|
"""Check if the provider is configured to be used."""
|
|
return True
|
|
|
|
def _get_cache_file_path(self, domain: str) -> Path:
|
|
"""Generate cache file path for a domain."""
|
|
safe_domain = domain.replace('.', '_').replace('/', '_').replace('\\', '_')
|
|
return self.cache_dir / f"{safe_domain}.json"
|
|
|
|
def _get_cache_status(self, cache_file_path: Path) -> str:
|
|
"""
|
|
Check cache status for a domain.
|
|
Returns: 'not_found', 'fresh', or 'stale'
|
|
"""
|
|
if not cache_file_path.exists():
|
|
return "not_found"
|
|
|
|
try:
|
|
with open(cache_file_path, 'r') as f:
|
|
cache_data = json.load(f)
|
|
|
|
last_query_str = cache_data.get("last_upstream_query")
|
|
if not last_query_str:
|
|
return "stale"
|
|
|
|
last_query = datetime.fromisoformat(last_query_str.replace('Z', '+00:00'))
|
|
hours_since_query = (datetime.now(timezone.utc) - last_query).total_seconds() / 3600
|
|
|
|
cache_timeout = self.config.cache_timeout_hours
|
|
if hours_since_query < cache_timeout:
|
|
return "fresh"
|
|
else:
|
|
return "stale"
|
|
|
|
except (json.JSONDecodeError, ValueError, KeyError) as e:
|
|
self.logger.logger.warning(f"Invalid cache file format for {cache_file_path}: {e}")
|
|
return "stale"
|
|
|
|
def query_domain(self, domain: str) -> ProviderResult:
|
|
"""
|
|
Query crt.sh for certificates containing the domain with efficient, deduplicated caching.
|
|
|
|
Args:
|
|
domain: Domain to investigate
|
|
|
|
Returns:
|
|
ProviderResult containing discovered relationships and attributes
|
|
"""
|
|
if not _is_valid_domain(domain):
|
|
return ProviderResult()
|
|
|
|
if self._stop_event and self._stop_event.is_set():
|
|
return ProviderResult()
|
|
|
|
cache_file = self._get_cache_file_path(domain)
|
|
cache_status = self._get_cache_status(cache_file)
|
|
|
|
result = ProviderResult()
|
|
|
|
try:
|
|
if cache_status == "fresh":
|
|
result = self._load_from_cache(cache_file)
|
|
#self.logger.logger.info(f"Using fresh cached crt.sh data for {domain}")
|
|
|
|
else: # "stale" or "not_found"
|
|
# Query the API for the latest certificates
|
|
new_raw_certs = self._query_crtsh_api(domain)
|
|
|
|
if self._stop_event and self._stop_event.is_set():
|
|
return ProviderResult()
|
|
|
|
# Combine with old data if cache is stale
|
|
if cache_status == "stale":
|
|
old_raw_certs = self._load_raw_data_from_cache(cache_file)
|
|
combined_certs = old_raw_certs + new_raw_certs
|
|
|
|
# Deduplicate the combined list
|
|
seen_ids = set()
|
|
unique_certs = []
|
|
for cert in combined_certs:
|
|
cert_id = cert.get('id')
|
|
if cert_id not in seen_ids:
|
|
unique_certs.append(cert)
|
|
seen_ids.add(cert_id)
|
|
|
|
raw_certificates_to_process = unique_certs
|
|
self.logger.logger.info(f"Refreshed and merged cache for {domain}. Total unique certs: {len(raw_certificates_to_process)}")
|
|
else: # "not_found"
|
|
raw_certificates_to_process = new_raw_certs
|
|
|
|
# Process the clean, deduplicated list of certificates
|
|
result = self._process_certificates_to_result(domain, raw_certificates_to_process)
|
|
self.logger.logger.info(f"Created fresh result for {domain} ({result.get_relationship_count()} relationships)")
|
|
|
|
# Save the new result and the raw data to the cache
|
|
self._save_result_to_cache(cache_file, result, raw_certificates_to_process, domain)
|
|
|
|
except requests.exceptions.RequestException as e:
|
|
self.logger.logger.error(f"API query failed for {domain}: {e}")
|
|
if cache_status != "not_found":
|
|
result = self._load_from_cache(cache_file)
|
|
self.logger.logger.warning(f"Using stale cache for {domain} due to API failure.")
|
|
else:
|
|
raise e # Re-raise if there's no cache to fall back on
|
|
|
|
return result
|
|
|
|
def query_ip(self, ip: str) -> ProviderResult:
|
|
"""
|
|
Query crt.sh for certificates containing the IP address.
|
|
Note: crt.sh doesn't typically index by IP, so this returns empty results.
|
|
|
|
Args:
|
|
ip: IP address to investigate
|
|
|
|
Returns:
|
|
Empty ProviderResult (crt.sh doesn't support IP-based certificate queries effectively)
|
|
"""
|
|
return ProviderResult()
|
|
|
|
def _load_from_cache(self, cache_file_path: Path) -> ProviderResult:
|
|
"""Load processed crt.sh data from a cache file."""
|
|
try:
|
|
with open(cache_file_path, 'r') as f:
|
|
cache_content = json.load(f)
|
|
|
|
result = ProviderResult()
|
|
|
|
# Reconstruct relationships
|
|
for rel_data in cache_content.get("relationships", []):
|
|
result.add_relationship(
|
|
source_node=rel_data["source_node"],
|
|
target_node=rel_data["target_node"],
|
|
relationship_type=rel_data["relationship_type"],
|
|
provider=rel_data["provider"],
|
|
confidence=rel_data["confidence"],
|
|
raw_data=rel_data.get("raw_data", {})
|
|
)
|
|
|
|
# Reconstruct attributes
|
|
for attr_data in cache_content.get("attributes", []):
|
|
result.add_attribute(
|
|
target_node=attr_data["target_node"],
|
|
name=attr_data["name"],
|
|
value=attr_data["value"],
|
|
attr_type=attr_data["type"],
|
|
provider=attr_data["provider"],
|
|
confidence=attr_data["confidence"],
|
|
metadata=attr_data.get("metadata", {})
|
|
)
|
|
|
|
return result
|
|
|
|
except (json.JSONDecodeError, FileNotFoundError, KeyError) as e:
|
|
self.logger.logger.error(f"Failed to load cached certificates from {cache_file_path}: {e}")
|
|
return ProviderResult()
|
|
|
|
def _load_raw_data_from_cache(self, cache_file_path: Path) -> List[Dict[str, Any]]:
|
|
"""Load only the raw certificate data from a cache file."""
|
|
try:
|
|
with open(cache_file_path, 'r') as f:
|
|
cache_content = json.load(f)
|
|
return cache_content.get("raw_certificates", [])
|
|
except (json.JSONDecodeError, FileNotFoundError):
|
|
return []
|
|
|
|
def _save_result_to_cache(self, cache_file_path: Path, result: ProviderResult, raw_certificates: List[Dict[str, Any]], domain: str) -> None:
|
|
"""Save processed crt.sh result and raw data to a cache file."""
|
|
try:
|
|
cache_data = {
|
|
"domain": domain,
|
|
"last_upstream_query": datetime.now(timezone.utc).isoformat(),
|
|
"raw_certificates": raw_certificates, # Store the raw data for deduplication
|
|
"relationships": [
|
|
{
|
|
"source_node": rel.source_node,
|
|
"target_node": rel.target_node,
|
|
"relationship_type": rel.relationship_type,
|
|
"confidence": rel.confidence,
|
|
"provider": rel.provider,
|
|
"raw_data": rel.raw_data
|
|
} for rel in result.relationships
|
|
],
|
|
"attributes": [
|
|
{
|
|
"target_node": attr.target_node,
|
|
"name": attr.name,
|
|
"value": attr.value,
|
|
"type": attr.type,
|
|
"provider": attr.provider,
|
|
"confidence": attr.confidence,
|
|
"metadata": attr.metadata
|
|
} for attr in result.attributes
|
|
]
|
|
}
|
|
cache_file_path.parent.mkdir(parents=True, exist_ok=True)
|
|
with open(cache_file_path, 'w') as f:
|
|
json.dump(cache_data, f, separators=(',', ':'), default=str)
|
|
except Exception as e:
|
|
self.logger.logger.warning(f"Failed to save cache file for {domain}: {e}")
|
|
|
|
def _query_crtsh_api(self, domain: str) -> List[Dict[str, Any]]:
|
|
"""Query crt.sh API for raw certificate data."""
|
|
url = f"{self.base_url}?q={quote(domain)}&output=json"
|
|
response = self.make_request(url, target_indicator=domain)
|
|
|
|
if not response or response.status_code != 200:
|
|
raise requests.exceptions.RequestException(f"crt.sh API returned status {response.status_code if response else 'None'}")
|
|
|
|
try:
|
|
certificates = response.json()
|
|
except json.JSONDecodeError:
|
|
self.logger.logger.error(f"crt.sh returned invalid JSON for {domain}")
|
|
return []
|
|
|
|
if not certificates:
|
|
return []
|
|
|
|
return certificates
|
|
|
|
def _process_certificates_to_result(self, domain: str, certificates: List[Dict[str, Any]]) -> ProviderResult:
|
|
"""
|
|
Process certificates to create ProviderResult with relationships and attributes.
|
|
"""
|
|
result = ProviderResult()
|
|
|
|
if self._stop_event and self._stop_event.is_set():
|
|
print(f"CrtSh processing cancelled before processing for domain: {domain}")
|
|
return result
|
|
|
|
all_discovered_domains = set()
|
|
|
|
for i, cert_data in enumerate(certificates):
|
|
if i % 5 == 0 and self._stop_event and self._stop_event.is_set():
|
|
print(f"CrtSh processing cancelled at certificate {i} for domain: {domain}")
|
|
break
|
|
|
|
cert_domains = self._extract_domains_from_certificate(cert_data)
|
|
all_discovered_domains.update(cert_domains)
|
|
|
|
for cert_domain in cert_domains:
|
|
if not _is_valid_domain(cert_domain):
|
|
continue
|
|
|
|
for key, value in self._extract_certificate_metadata(cert_data).items():
|
|
if value is not None:
|
|
result.add_attribute(
|
|
target_node=cert_domain,
|
|
name=f"cert_{key}",
|
|
value=value,
|
|
attr_type='certificate_data',
|
|
provider=self.name,
|
|
confidence=0.9
|
|
)
|
|
|
|
if self._stop_event and self._stop_event.is_set():
|
|
print(f"CrtSh query cancelled before relationship creation for domain: {domain}")
|
|
return result
|
|
|
|
for i, discovered_domain in enumerate(all_discovered_domains):
|
|
if discovered_domain == domain:
|
|
continue
|
|
|
|
if i % 10 == 0 and self._stop_event and self._stop_event.is_set():
|
|
print(f"CrtSh relationship creation cancelled for domain: {domain}")
|
|
break
|
|
|
|
if not _is_valid_domain(discovered_domain):
|
|
continue
|
|
|
|
confidence = self._calculate_domain_relationship_confidence(
|
|
domain, discovered_domain, [], all_discovered_domains
|
|
)
|
|
|
|
result.add_relationship(
|
|
source_node=domain,
|
|
target_node=discovered_domain,
|
|
relationship_type='crtsh_san_certificate',
|
|
provider=self.name,
|
|
confidence=confidence,
|
|
raw_data={'relationship_type': 'certificate_discovery'}
|
|
)
|
|
|
|
self.log_relationship_discovery(
|
|
source_node=domain,
|
|
target_node=discovered_domain,
|
|
relationship_type='crtsh_san_certificate',
|
|
confidence_score=confidence,
|
|
raw_data={'relationship_type': 'certificate_discovery'},
|
|
discovery_method="certificate_transparency_analysis"
|
|
)
|
|
|
|
return result
|
|
|
|
def _extract_certificate_metadata(self, cert_data: Dict[str, Any]) -> Dict[str, Any]:
|
|
"""Extract comprehensive metadata from certificate data."""
|
|
raw_issuer_name = cert_data.get('issuer_name', '')
|
|
parsed_issuer_name = self._parse_issuer_organization(raw_issuer_name)
|
|
|
|
metadata = {
|
|
'certificate_id': cert_data.get('id'),
|
|
'serial_number': cert_data.get('serial_number'),
|
|
'issuer_name': parsed_issuer_name,
|
|
'issuer_ca_id': cert_data.get('issuer_ca_id'),
|
|
'common_name': cert_data.get('common_name'),
|
|
'not_before': cert_data.get('not_before'),
|
|
'not_after': cert_data.get('not_after'),
|
|
'entry_timestamp': cert_data.get('entry_timestamp'),
|
|
'source': 'crtsh'
|
|
}
|
|
|
|
try:
|
|
if metadata['not_before'] and metadata['not_after']:
|
|
not_before = self._parse_certificate_date(metadata['not_before'])
|
|
not_after = self._parse_certificate_date(metadata['not_after'])
|
|
|
|
metadata['validity_period_days'] = (not_after - not_before).days
|
|
metadata['is_currently_valid'] = self._is_cert_valid(cert_data)
|
|
metadata['expires_soon'] = (not_after - datetime.now(timezone.utc)).days <= 30
|
|
|
|
# UPDATED: Keep raw date format or convert to standard format
|
|
metadata['not_before'] = not_before.isoformat()
|
|
metadata['not_after'] = not_after.isoformat()
|
|
|
|
except Exception as e:
|
|
self.logger.logger.debug(f"Error computing certificate metadata: {e}")
|
|
metadata['is_currently_valid'] = False
|
|
metadata['expires_soon'] = False
|
|
|
|
return metadata
|
|
|
|
def _parse_issuer_organization(self, issuer_dn: str) -> str:
|
|
"""Parse the issuer Distinguished Name to extract just the organization name."""
|
|
if not issuer_dn:
|
|
return issuer_dn
|
|
|
|
try:
|
|
components = [comp.strip() for comp in issuer_dn.split(',')]
|
|
|
|
for component in components:
|
|
if component.startswith('O='):
|
|
org_name = component[2:].strip()
|
|
if org_name.startswith('"') and org_name.endswith('"'):
|
|
org_name = org_name[1:-1]
|
|
return org_name
|
|
|
|
return issuer_dn
|
|
|
|
except Exception as e:
|
|
self.logger.logger.debug(f"Failed to parse issuer DN '{issuer_dn}': {e}")
|
|
return issuer_dn
|
|
|
|
def _parse_certificate_date(self, date_string: str) -> datetime:
|
|
"""Parse certificate date from crt.sh format."""
|
|
if not date_string:
|
|
raise ValueError("Empty date string")
|
|
|
|
try:
|
|
if date_string.endswith('Z'):
|
|
return datetime.fromisoformat(date_string[:-1]).replace(tzinfo=timezone.utc)
|
|
elif '+' in date_string or date_string.endswith('UTC'):
|
|
date_string = date_string.replace('UTC', '').strip()
|
|
if '+' in date_string:
|
|
date_string = date_string.split('+')[0]
|
|
return datetime.fromisoformat(date_string).replace(tzinfo=timezone.utc)
|
|
else:
|
|
return datetime.fromisoformat(date_string).replace(tzinfo=timezone.utc)
|
|
except Exception as e:
|
|
try:
|
|
return datetime.strptime(date_string[:19], "%Y-%m-%dT%H:%M:%S").replace(tzinfo=timezone.utc)
|
|
except Exception:
|
|
raise ValueError(f"Unable to parse date: {date_string}") from e
|
|
|
|
def _is_cert_valid(self, cert_data: Dict[str, Any]) -> bool:
|
|
"""Check if a certificate is currently valid based on its expiry date."""
|
|
try:
|
|
not_after_str = cert_data.get('not_after')
|
|
if not not_after_str:
|
|
return False
|
|
|
|
not_after_date = self._parse_certificate_date(not_after_str)
|
|
not_before_str = cert_data.get('not_before')
|
|
|
|
now = datetime.now(timezone.utc)
|
|
is_not_expired = not_after_date > now
|
|
|
|
if not_before_str:
|
|
not_before_date = self._parse_certificate_date(not_before_str)
|
|
is_not_before_valid = not_before_date <= now
|
|
return is_not_expired and is_not_before_valid
|
|
|
|
return is_not_expired
|
|
|
|
except Exception as e:
|
|
#self.logger.logger.debug(f"Certificate validity check failed: {e}")
|
|
return False
|
|
|
|
def _extract_domains_from_certificate(self, cert_data: Dict[str, Any]) -> Set[str]:
|
|
"""Extract all domains from certificate data."""
|
|
domains = set()
|
|
|
|
# Extract from common name
|
|
common_name = cert_data.get('common_name', '')
|
|
if common_name:
|
|
cleaned_cn = self._clean_domain_name(common_name)
|
|
if cleaned_cn:
|
|
domains.update(cleaned_cn)
|
|
|
|
# Extract from name_value field (contains SANs)
|
|
name_value = cert_data.get('name_value', '')
|
|
if name_value:
|
|
for line in name_value.split('\n'):
|
|
cleaned_domains = self._clean_domain_name(line.strip())
|
|
if cleaned_domains:
|
|
domains.update(cleaned_domains)
|
|
|
|
return domains
|
|
|
|
def _clean_domain_name(self, domain_name: str) -> List[str]:
|
|
"""Clean and normalize domain name from certificate data."""
|
|
if not domain_name:
|
|
return []
|
|
|
|
domain = domain_name.strip().lower()
|
|
|
|
if domain.startswith(('http://', 'https://')):
|
|
domain = domain.split('://', 1)[1]
|
|
|
|
if '/' in domain:
|
|
domain = domain.split('/', 1)[0]
|
|
|
|
if ':' in domain and not domain.count(':') > 1:
|
|
domain = domain.split(':', 1)[0]
|
|
|
|
cleaned_domains = []
|
|
if domain.startswith('*.'):
|
|
cleaned_domains.append(domain)
|
|
cleaned_domains.append(domain[2:])
|
|
else:
|
|
cleaned_domains.append(domain)
|
|
|
|
final_domains = []
|
|
for d in cleaned_domains:
|
|
d = re.sub(r'[^\w\-\.]', '', d)
|
|
if d and not d.startswith(('.', '-')) and not d.endswith(('.', '-')):
|
|
final_domains.append(d)
|
|
|
|
return [d for d in final_domains if _is_valid_domain(d)]
|
|
|
|
def _get_certificate_sort_date(self, cert: Dict[str, Any]) -> datetime:
|
|
"""Get a sortable date from certificate data for chronological ordering."""
|
|
try:
|
|
if cert.get('not_before'):
|
|
return self._parse_certificate_date(cert['not_before'])
|
|
|
|
if cert.get('entry_timestamp'):
|
|
return self._parse_certificate_date(cert['entry_timestamp'])
|
|
|
|
return datetime(1970, 1, 1, tzinfo=timezone.utc)
|
|
|
|
except Exception:
|
|
return datetime(1970, 1, 1, tzinfo=timezone.utc)
|
|
|
|
def _calculate_domain_relationship_confidence(self, domain1: str, domain2: str,
|
|
shared_certificates: List[Dict[str, Any]],
|
|
all_discovered_domains: Set[str]) -> float:
|
|
"""Calculate confidence score for domain relationship based on various factors."""
|
|
base_confidence = 0.9
|
|
|
|
# Adjust confidence based on domain relationship context
|
|
relationship_context = self._determine_relationship_context(domain2, domain1)
|
|
|
|
if relationship_context == 'exact_match':
|
|
context_bonus = 0.0
|
|
elif relationship_context == 'subdomain':
|
|
context_bonus = 0.1
|
|
elif relationship_context == 'parent_domain':
|
|
context_bonus = 0.05
|
|
else:
|
|
context_bonus = 0.0
|
|
|
|
# Adjust confidence based on shared certificates
|
|
if shared_certificates:
|
|
shared_count = len(shared_certificates)
|
|
if shared_count >= 3:
|
|
shared_bonus = 0.1
|
|
elif shared_count >= 2:
|
|
shared_bonus = 0.05
|
|
else:
|
|
shared_bonus = 0.02
|
|
|
|
valid_shared = sum(1 for cert in shared_certificates if cert.get('is_currently_valid'))
|
|
if valid_shared > 0:
|
|
validity_bonus = 0.05
|
|
else:
|
|
validity_bonus = 0.0
|
|
else:
|
|
shared_bonus = 0.0
|
|
validity_bonus = 0.0
|
|
|
|
# Adjust confidence based on certificate issuer reputation
|
|
issuer_bonus = 0.0
|
|
if shared_certificates:
|
|
for cert in shared_certificates:
|
|
issuer = cert.get('issuer_name', '').lower()
|
|
if any(trusted_ca in issuer for trusted_ca in ['let\'s encrypt', 'digicert', 'sectigo', 'globalsign']):
|
|
issuer_bonus = max(issuer_bonus, 0.03)
|
|
break
|
|
|
|
final_confidence = base_confidence + context_bonus + shared_bonus + validity_bonus + issuer_bonus
|
|
return max(0.1, min(1.0, final_confidence))
|
|
|
|
def _determine_relationship_context(self, cert_domain: str, query_domain: str) -> str:
|
|
"""Determine the context of the relationship between certificate domain and query domain."""
|
|
if cert_domain == query_domain:
|
|
return 'exact_match'
|
|
elif cert_domain.endswith(f'.{query_domain}'):
|
|
return 'subdomain'
|
|
elif query_domain.endswith(f'.{cert_domain}'):
|
|
return 'parent_domain'
|
|
else:
|
|
return 'related_domain' |