mstoeck3/dnscope
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

upgrades

Browse Source
This commit is contained in:
overcuriousity 2025-09-10 16:26:44 +02:00
parent ce0e11cf0b
commit a0caedcb1f
4 changed files with 122 additions and 44 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
core/graph_manager.py
View File

@ -102,8 +102,10 @@ class GraphManager:
#with self.lock:
# Ensure both nodes exist
if not self.graph.has_node(source_id) or not self.graph.has_node(target_id):
return False
# If the target node is a subdomain, it should be added.
# The scanner will handle this logic.
pass
# Check if edge already exists
if self.graph.has_edge(source_id, target_id):
# Update confidence score if new score is higher
@ -241,6 +243,11 @@ class GraphManager:
node_color_config = type_colors.get(attributes.get('type', 'unknown'), type_colors['domain'])
node_data['color'] = node_color_config
# Pass the has_valid_cert metadata for styling
if 'metadata' in attributes and 'has_valid_cert' in attributes['metadata']:
node_data['has_valid_cert'] = attributes['metadata']['has_valid_cert']
nodes.append(node_data)
# Format edges for visualization

51
core/scanner.py
View File

@ -626,6 +626,57 @@ class Scanner:
for provider in self.providers:
stats[provider.get_name()] = provider.get_statistics()
return stats
def _is_valid_domain(self, domain: str) -> bool:
"""
Basic domain validation.
Args:
domain: Domain string to validate
Returns:
True if domain appears valid
"""
if not domain or len(domain) > 253:
return False
# Check for valid characters and structure
parts = domain.split('.')
if len(parts) < 2:
return False
for part in parts:
if not part or len(part) > 63:
return False
if not part.replace('-', '').replace('_', '').isalnum():
return False
return True
def _is_valid_ip(self, ip: str) -> bool:
"""
Basic IP address validation.
Args:
ip: IP address string to validate
Returns:
True if IP appears valid
"""
try:
parts = ip.split('.')
if len(parts) != 4:
return False
for part in parts:
num = int(part)
if not 0 <= num <= 255:
return False
return True
except (ValueError, AttributeError):
return False
class ScannerProxy:

96
providers/crtsh_provider.py
View File

@ -7,6 +7,7 @@ import json
import re
from typing import List, Dict, Any, Tuple, Set
from urllib.parse import quote
from datetime import datetime, timezone
from .base_provider import BaseProvider
from core.graph_manager import RelationshipType
@ -39,6 +40,20 @@ class CrtShProvider(BaseProvider):
"""
return True
def _is_cert_valid(self, cert_data: Dict[str, Any]) -> bool:
"""Check if a certificate is currently valid."""
try:
not_after_str = cert_data.get('not_after')
if not_after_str:
# Append 'Z' to indicate UTC if it's not present
if not not_after_str.endswith('Z'):
not_after_str += 'Z'
not_after_date = datetime.fromisoformat(not_after_str.replace('Z', '+00:00'))
return not_after_date > datetime.now(timezone.utc)
except Exception:
return False
return False
def query_domain(self, domain: str) -> List[Tuple[str, str, RelationshipType, float, Dict[str, Any]]]:
"""
Query crt.sh for certificates containing the domain.
@ -68,50 +83,47 @@ class CrtShProvider(BaseProvider):
return []
# Process certificates to extract relationships
seen_certificates = set()
discovered_subdomains = {}
for cert_data in certificates:
cert_id = cert_data.get('id')
if not cert_id or cert_id in seen_certificates:
continue
seen_certificates.add(cert_id)
# Extract domains from certificate
cert_domains = self._extract_domains_from_certificate(cert_data)
if domain in cert_domains and len(cert_domains) > 1:
# Create relationships between domains found in the same certificate
for related_domain in cert_domains:
if related_domain != domain and self._is_valid_domain(related_domain):
# Create SAN relationship
raw_data = {
'certificate_id': cert_id,
'issuer': cert_data.get('issuer_name', ''),
'not_before': cert_data.get('not_before', ''),
'not_after': cert_data.get('not_after', ''),
'serial_number': cert_data.get('serial_number', ''),
'all_domains': list(cert_domains)
}
relationships.append((
domain,
related_domain,
RelationshipType.SAN_CERTIFICATE,
RelationshipType.SAN_CERTIFICATE.default_confidence,
raw_data
))
# Log the discovery
self.log_relationship_discovery(
source_node=domain,
target_node=related_domain,
relationship_type=RelationshipType.SAN_CERTIFICATE,
confidence_score=RelationshipType.SAN_CERTIFICATE.default_confidence,
raw_data=raw_data,
discovery_method="certificate_san_analysis"
)
is_valid = self._is_cert_valid(cert_data)
for subdomain in cert_domains:
if self._is_valid_domain(subdomain) and subdomain != domain:
if subdomain not in discovered_subdomains:
discovered_subdomains[subdomain] = {'has_valid_cert': False, 'issuers': set()}
if is_valid:
discovered_subdomains[subdomain]['has_valid_cert'] = True
issuer = cert_data.get('issuer_name')
if issuer:
discovered_subdomains[subdomain]['issuers'].add(issuer)
# Create relationships from the discovered subdomains
for subdomain, data in discovered_subdomains.items():
raw_data = {
'has_valid_cert': data['has_valid_cert'],
'issuers': list(data['issuers']),
'source': 'crt.sh'
}
relationships.append((
domain,
subdomain,
RelationshipType.SAN_CERTIFICATE,
RelationshipType.SAN_CERTIFICATE.default_confidence,
raw_data
))
self.log_relationship_discovery(
source_node=domain,
target_node=subdomain,
relationship_type=RelationshipType.SAN_CERTIFICATE,
confidence_score=RelationshipType.SAN_CERTIFICATE.default_confidence,
raw_data=raw_data,
discovery_method="certificate_san_analysis"
)
except json.JSONDecodeError as e:
self.logger.logger.error(f"Failed to parse JSON response from crt.sh: {e}")
except Exception as e:

8
static/js/graph.js
View File

@ -366,6 +366,14 @@ class GraphManager {
};
}
// Style based on certificate validity
if (node.has_valid_cert === true) {
processedNode.borderColor = '#00ff41'; // Green for valid cert
} else if (node.has_valid_cert === false) {
processedNode.borderColor = '#ff9900'; // Amber for expired/no cert
processedNode.borderDashes = [5, 5];
}
return processedNode;
}