# DNScope/providers/crtsh_provider.py import json import re from pathlib import Path from typing import List, Dict, Any, Set, Optional 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 from core.logger import get_forensic_logger class CrtShProvider(BaseProvider): """ Provider for querying crt.sh certificate transparency database. FIXED: Improved caching logic and error handling to prevent infinite retry loops. 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.domain_cache_dir = Path('cache') / 'crtsh' self.domain_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.domain_cache_dir / f"{safe_domain}.json" def _get_cache_status(self, cache_file_path: Path) -> str: """ FIXED: More robust cache status checking with better error handling. Returns: 'not_found', 'fresh', or 'stale' """ if not cache_file_path.exists(): return "not_found" try: # Check if file is readable and not corrupted if cache_file_path.stat().st_size == 0: self.logger.logger.warning(f"Empty cache file: {cache_file_path}") return "stale" with open(cache_file_path, 'r', encoding='utf-8') as f: cache_data = json.load(f) # Validate cache structure if not isinstance(cache_data, dict): self.logger.logger.warning(f"Invalid cache structure: {cache_file_path}") return "stale" last_query_str = cache_data.get("last_upstream_query") if not last_query_str or not isinstance(last_query_str, str): self.logger.logger.warning(f"Missing or invalid last_upstream_query: {cache_file_path}") return "stale" try: # More robust datetime parsing if last_query_str.endswith('Z'): last_query = datetime.fromisoformat(last_query_str.replace('Z', '+00:00')) elif '+' in last_query_str or last_query_str.endswith('UTC'): # Handle various timezone formats clean_time = last_query_str.replace('UTC', '').strip() if '+' in clean_time: clean_time = clean_time.split('+')[0] last_query = datetime.fromisoformat(clean_time).replace(tzinfo=timezone.utc) else: last_query = datetime.fromisoformat(last_query_str).replace(tzinfo=timezone.utc) except (ValueError, AttributeError) as e: self.logger.logger.warning(f"Failed to parse timestamp in cache {cache_file_path}: {e}") return "stale" 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, OSError, PermissionError) as e: self.logger.logger.warning(f"Cache file error for {cache_file_path}: {e}") # FIXED: Try to remove corrupted cache file try: cache_file_path.unlink() self.logger.logger.info(f"Removed corrupted cache file: {cache_file_path}") except Exception: pass return "not_found" except Exception as e: self.logger.logger.error(f"Unexpected error checking cache status for {cache_file_path}: {e}") return "stale" def query_domain(self, domain: str) -> ProviderResult: """ FIXED: Simplified and more robust domain querying with better error handling. """ 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) result = ProviderResult() try: cache_status = self._get_cache_status(cache_file) if cache_status == "fresh": # Load from cache result = self._load_from_cache(cache_file) if result and (result.relationships or result.attributes): self.logger.logger.debug(f"Using fresh cached crt.sh data for {domain}") return result else: # Cache exists but is empty, treat as stale cache_status = "stale" # Need to query API (either no cache, stale cache, or empty cache) self.logger.logger.debug(f"Querying crt.sh API for {domain} (cache status: {cache_status})") new_raw_certs = self._query_crtsh_api(domain) if self._stop_event and self._stop_event.is_set(): return ProviderResult() # FIXED: Simplified processing - just process the new data # Don't try to merge with stale cache as it can cause corruption raw_certificates_to_process = new_raw_certs if cache_status == "stale": self.logger.logger.info(f"Refreshed stale cache for {domain} with {len(raw_certificates_to_process)} certs") else: self.logger.logger.info(f"Created fresh cache for {domain} with {len(raw_certificates_to_process)} certs") result = self._process_certificates_to_result_fixed(domain, raw_certificates_to_process) # Save the result to cache self._save_result_to_cache(cache_file, result, raw_certificates_to_process, domain) return result except requests.exceptions.RequestException as e: # FIXED: Don't re-raise network errors after long idle periods # Instead return empty result and log the issue self.logger.logger.warning(f"Network error querying crt.sh for {domain}: {e}") # Try to use stale cache if available if cache_status == "stale": try: stale_result = self._load_from_cache(cache_file) if stale_result and (stale_result.relationships or stale_result.attributes): self.logger.logger.info(f"Using stale cache for {domain} due to network error") return stale_result except Exception as cache_error: self.logger.logger.warning(f"Failed to load stale cache for {domain}: {cache_error}") # Return empty result instead of raising - this prevents infinite retries return ProviderResult() except Exception as e: # FIXED: Handle any other exceptions gracefully self.logger.logger.error(f"Unexpected error querying crt.sh for {domain}: {e}") # Try stale cache as fallback try: if cache_file.exists(): fallback_result = self._load_from_cache(cache_file) if fallback_result and (fallback_result.relationships or fallback_result.attributes): self.logger.logger.info(f"Using cached data for {domain} due to processing error") return fallback_result except Exception: pass # Return empty result to prevent retries return ProviderResult() def query_ip(self, ip: str) -> ProviderResult: """ crt.sh does not support IP-based certificate queries effectively via its API. """ return ProviderResult() def _load_from_cache(self, cache_file_path: Path) -> ProviderResult: """FIXED: More robust cache loading with better validation.""" try: if not cache_file_path.exists() or cache_file_path.stat().st_size == 0: return ProviderResult() with open(cache_file_path, 'r', encoding='utf-8') as f: cache_content = json.load(f) if not isinstance(cache_content, dict): self.logger.logger.warning(f"Invalid cache format in {cache_file_path}") return ProviderResult() result = ProviderResult() # Reconstruct relationships with validation relationships = cache_content.get("relationships", []) if isinstance(relationships, list): for rel_data in relationships: if not isinstance(rel_data, dict): continue try: result.add_relationship( source_node=rel_data.get("source_node", ""), target_node=rel_data.get("target_node", ""), relationship_type=rel_data.get("relationship_type", ""), provider=rel_data.get("provider", self.name), confidence=float(rel_data.get("confidence", 0.8)), raw_data=rel_data.get("raw_data", {}) ) except (ValueError, TypeError) as e: self.logger.logger.warning(f"Skipping invalid relationship in cache: {e}") continue # Reconstruct attributes with validation attributes = cache_content.get("attributes", []) if isinstance(attributes, list): for attr_data in attributes: if not isinstance(attr_data, dict): continue try: result.add_attribute( target_node=attr_data.get("target_node", ""), name=attr_data.get("name", ""), value=attr_data.get("value"), attr_type=attr_data.get("type", "unknown"), provider=attr_data.get("provider", self.name), confidence=float(attr_data.get("confidence", 0.9)), metadata=attr_data.get("metadata", {}) ) except (ValueError, TypeError) as e: self.logger.logger.warning(f"Skipping invalid attribute in cache: {e}") continue return result except (json.JSONDecodeError, OSError, PermissionError) as e: self.logger.logger.warning(f"Failed to load cache from {cache_file_path}: {e}") return ProviderResult() except Exception as e: self.logger.logger.error(f"Unexpected error loading cache from {cache_file_path}: {e}") return ProviderResult() def _save_result_to_cache(self, cache_file_path: Path, result: ProviderResult, raw_certificates: List[Dict[str, Any]], domain: str) -> None: """FIXED: More robust cache saving with atomic writes.""" try: cache_data = { "domain": domain, "last_upstream_query": datetime.now(timezone.utc).isoformat(), "raw_certificates": raw_certificates, "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) # FIXED: Atomic write using temporary file temp_file = cache_file_path.with_suffix('.tmp') try: with open(temp_file, 'w', encoding='utf-8') as f: json.dump(cache_data, f, separators=(',', ':'), default=str, ensure_ascii=False) # Atomic rename temp_file.replace(cache_file_path) self.logger.logger.debug(f"Saved cache for {domain} ({len(result.relationships)} relationships)") except Exception as e: # Clean up temp file on error if temp_file.exists(): try: temp_file.unlink() except Exception: pass raise e 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]]: """FIXED: More robust API querying with better error handling.""" url = f"{self.base_url}?q={quote(domain)}&output=json" try: response = self.make_request(url, target_indicator=domain) if not response: self.logger.logger.warning(f"No response from crt.sh for {domain}") return [] if response.status_code != 200: self.logger.logger.warning(f"crt.sh returned status {response.status_code} for {domain}") return [] # FIXED: Better JSON parsing with error handling try: certificates = response.json() except json.JSONDecodeError as e: self.logger.logger.error(f"crt.sh returned invalid JSON for {domain}: {e}") return [] if not certificates or not isinstance(certificates, list): self.logger.logger.debug(f"crt.sh returned no certificates for {domain}") return [] self.logger.logger.debug(f"crt.sh returned {len(certificates)} certificates for {domain}") return certificates except Exception as e: self.logger.logger.error(f"Error querying crt.sh API for {domain}: {e}") raise e def _process_certificates_to_result_fixed(self, query_domain: str, certificates: List[Dict[str, Any]]) -> ProviderResult: """ Process certificates to create proper domain and CA nodes. FIXED: Better error handling and progress tracking. """ result = ProviderResult() if self._stop_event and self._stop_event.is_set(): self.logger.logger.info(f"CrtSh processing cancelled before processing for domain: {query_domain}") return result if not certificates: self.logger.logger.debug(f"No certificates to process for {query_domain}") return result # Check for incomplete data warning incompleteness_warning = self._check_for_incomplete_data(query_domain, certificates) if incompleteness_warning: result.add_attribute( target_node=query_domain, name="crtsh_data_warning", value=incompleteness_warning, attr_type='metadata', provider=self.name, confidence=1.0 ) all_discovered_domains = set() processed_issuers = set() processed_certs = 0 for i, cert_data in enumerate(certificates): # FIXED: More frequent stop checks and progress logging if i % 5 == 0: if self._stop_event and self._stop_event.is_set(): self.logger.logger.info(f"CrtSh processing cancelled at certificate {i}/{len(certificates)} for domain: {query_domain}") break if i > 0 and i % 100 == 0: self.logger.logger.debug(f"Processed {i}/{len(certificates)} certificates for {query_domain}") try: # Extract all domains from this certificate cert_domains = self._extract_domains_from_certificate(cert_data) if cert_domains: all_discovered_domains.update(cert_domains) # Create CA nodes for certificate issuers issuer_name = self._parse_issuer_organization(cert_data.get('issuer_name', '')) if issuer_name and issuer_name not in processed_issuers: result.add_relationship( source_node=query_domain, target_node=issuer_name, relationship_type='crtsh_cert_issuer', provider=self.name, confidence=0.95, raw_data={'issuer_dn': cert_data.get('issuer_name', '')} ) processed_issuers.add(issuer_name) # Add certificate metadata to each domain in this certificate cert_metadata = self._extract_certificate_metadata(cert_data) for cert_domain in cert_domains: if not _is_valid_domain(cert_domain): continue for key, value in cert_metadata.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, metadata={'certificate_id': cert_data.get('id')} ) processed_certs += 1 except Exception as e: self.logger.logger.warning(f"Error processing certificate {i} for {query_domain}: {e}") continue # Check for stop event before creating final relationships if self._stop_event and self._stop_event.is_set(): self.logger.logger.info(f"CrtSh query cancelled before relationship creation for domain: {query_domain}") return result # Create selective relationships to avoid large entities relationships_created = 0 for discovered_domain in all_discovered_domains: if discovered_domain == query_domain: continue if not _is_valid_domain(discovered_domain): continue if self._should_create_relationship(query_domain, discovered_domain): confidence = self._calculate_domain_relationship_confidence( query_domain, discovered_domain, [], all_discovered_domains ) result.add_relationship( source_node=query_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=query_domain, target_node=discovered_domain, relationship_type='crtsh_san_certificate', confidence_score=confidence, raw_data={'relationship_type': 'certificate_discovery'}, discovery_method="certificate_transparency_analysis" ) relationships_created += 1 self.logger.logger.info(f"CrtSh processing completed for {query_domain}: processed {processed_certs}/{len(certificates)} certificates, {len(all_discovered_domains)} domains, {relationships_created} relationships") return result # [Rest of the methods remain the same as in the original file] def _should_create_relationship(self, source_domain: str, target_domain: str) -> bool: """ Determine if a relationship should be created between two domains. """ if target_domain.endswith(f'.{source_domain}') or source_domain.endswith(f'.{target_domain}'): return True source_parts = source_domain.split('.') target_parts = target_domain.split('.') if len(source_parts) >= 2 and len(target_parts) >= 2: source_root = '.'.join(source_parts[-2:]) target_root = '.'.join(target_parts[-2:]) return source_root == target_root return False 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 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 isinstance(date_string, datetime): return date_string.replace(tzinfo=timezone.utc) 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: return False def _extract_domains_from_certificate(self, cert_data: Dict[str, Any]) -> Set[str]: """Extract all domains from certificate data.""" domains = set() 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) 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 _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 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 final_confidence = base_confidence + context_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' def _check_for_incomplete_data(self, domain: str, certificates: List[Dict[str, Any]]) -> Optional[str]: """ Analyzes the certificate list to heuristically detect if the data from crt.sh is incomplete. """ cert_count = len(certificates) if cert_count >= 10000: return f"Result likely truncated; received {cert_count} certificates, which may be the maximum limit." if cert_count > 1000: latest_expiry = None for cert in certificates: try: not_after = self._parse_certificate_date(cert.get('not_after')) if latest_expiry is None or not_after > latest_expiry: latest_expiry = not_after except (ValueError, TypeError): continue if latest_expiry and (datetime.now(timezone.utc) - latest_expiry).days > 365: return f"Incomplete data suspected: The latest certificate expired more than a year ago ({latest_expiry.strftime('%Y-%m-%d')})." return None