Phillip Tarrant
66b02edc84
Save screenshot_dir to database when scans complete so the directory is properly cleaned up on scan deletion. Previously the field was never populated, causing screenshots to remain after deleting scans. Update sslyze to 6.2.0 and cryptography to 46.0.0 to fix certificate handling issues with negative serial numbers (RFC 5280 compliance). 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <noreply@anthropic.com>
1050 lines
38 KiB
Python
1050 lines
38 KiB
Python
"""
|
|
Scan service for managing scan operations and database integration.
|
|
|
|
This service handles the business logic for triggering scans, retrieving
|
|
scan results, and mapping scanner output to database models.
|
|
"""
|
|
|
|
import json
|
|
import logging
|
|
import shutil
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
from sqlalchemy.orm import Session, joinedload
|
|
|
|
from web.models import (
|
|
Scan, ScanSite, ScanIP, ScanPort, ScanService as ScanServiceModel,
|
|
ScanCertificate, ScanTLSVersion, Site, ScanSiteAssociation
|
|
)
|
|
from web.utils.pagination import paginate, PaginatedResult
|
|
from web.utils.validators import validate_scan_status
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class ScanService:
|
|
"""
|
|
Service for managing scan operations.
|
|
|
|
Handles scan lifecycle: triggering, status tracking, result storage,
|
|
and cleanup.
|
|
"""
|
|
|
|
def __init__(self, db_session: Session):
|
|
"""
|
|
Initialize scan service.
|
|
|
|
Args:
|
|
db_session: SQLAlchemy database session
|
|
"""
|
|
self.db = db_session
|
|
|
|
def trigger_scan(self, config_id: int,
|
|
triggered_by: str = 'manual', schedule_id: Optional[int] = None,
|
|
scheduler=None) -> int:
|
|
"""
|
|
Trigger a new scan.
|
|
|
|
Creates a Scan record in the database with status='running' and
|
|
queues the scan for background execution.
|
|
|
|
Args:
|
|
config_id: Database config ID
|
|
triggered_by: Source that triggered scan (manual, scheduled, api)
|
|
schedule_id: Optional schedule ID if triggered by schedule
|
|
scheduler: Optional SchedulerService instance for queuing background jobs
|
|
|
|
Returns:
|
|
Scan ID of the created scan
|
|
|
|
Raises:
|
|
ValueError: If config is invalid
|
|
"""
|
|
from web.models import ScanConfig
|
|
|
|
# Validate config exists
|
|
db_config = self.db.query(ScanConfig).filter_by(id=config_id).first()
|
|
if not db_config:
|
|
raise ValueError(f"Config with ID {config_id} not found")
|
|
|
|
# Create scan record with config_id
|
|
scan = Scan(
|
|
timestamp=datetime.utcnow(),
|
|
status='running',
|
|
config_id=config_id,
|
|
title=db_config.title,
|
|
triggered_by=triggered_by,
|
|
schedule_id=schedule_id,
|
|
created_at=datetime.utcnow()
|
|
)
|
|
|
|
self.db.add(scan)
|
|
self.db.commit()
|
|
self.db.refresh(scan)
|
|
|
|
logger.info(f"Scan {scan.id} triggered via {triggered_by} with config_id={config_id}")
|
|
|
|
# Queue background job if scheduler provided
|
|
if scheduler:
|
|
try:
|
|
job_id = scheduler.queue_scan(scan.id, config_id=config_id)
|
|
logger.info(f"Scan {scan.id} queued for background execution (job_id={job_id})")
|
|
except Exception as e:
|
|
logger.error(f"Failed to queue scan {scan.id}: {str(e)}")
|
|
# Mark scan as failed if job queuing fails
|
|
scan.status = 'failed'
|
|
scan.error_message = f"Failed to queue background job: {str(e)}"
|
|
self.db.commit()
|
|
raise
|
|
else:
|
|
logger.warning(f"Scan {scan.id} created but not queued (no scheduler provided)")
|
|
|
|
return scan.id
|
|
|
|
def get_scan(self, scan_id: int) -> Optional[Dict[str, Any]]:
|
|
"""
|
|
Get scan details with all related data.
|
|
|
|
Args:
|
|
scan_id: Scan ID to retrieve
|
|
|
|
Returns:
|
|
Dictionary with scan data including sites, IPs, ports, services, etc.
|
|
Returns None if scan not found.
|
|
"""
|
|
# Query with eager loading of all relationships
|
|
scan = (
|
|
self.db.query(Scan)
|
|
.options(
|
|
joinedload(Scan.sites).joinedload(ScanSite.ips).joinedload(ScanIP.ports),
|
|
joinedload(Scan.ports).joinedload(ScanPort.services),
|
|
joinedload(Scan.services).joinedload(ScanServiceModel.certificates),
|
|
joinedload(Scan.certificates).joinedload(ScanCertificate.tls_versions)
|
|
)
|
|
.filter(Scan.id == scan_id)
|
|
.first()
|
|
)
|
|
|
|
if not scan:
|
|
return None
|
|
|
|
# Convert to dictionary
|
|
return self._scan_to_dict(scan)
|
|
|
|
def list_scans(self, page: int = 1, per_page: int = 20,
|
|
status_filter: Optional[str] = None) -> PaginatedResult:
|
|
"""
|
|
List scans with pagination and optional filtering.
|
|
|
|
Args:
|
|
page: Page number (1-indexed)
|
|
per_page: Items per page
|
|
status_filter: Optional filter by status (running, completed, failed)
|
|
|
|
Returns:
|
|
PaginatedResult with scan list and metadata
|
|
"""
|
|
# Build query
|
|
query = self.db.query(Scan).order_by(Scan.timestamp.desc())
|
|
|
|
# Apply status filter if provided
|
|
if status_filter:
|
|
is_valid, error_msg = validate_scan_status(status_filter)
|
|
if not is_valid:
|
|
raise ValueError(error_msg)
|
|
query = query.filter(Scan.status == status_filter)
|
|
|
|
# Paginate
|
|
result = paginate(query, page=page, per_page=per_page)
|
|
|
|
# Convert scans to dictionaries (summary only, not full details)
|
|
result.items = [self._scan_to_summary_dict(scan) for scan in result.items]
|
|
|
|
return result
|
|
|
|
def delete_scan(self, scan_id: int) -> bool:
|
|
"""
|
|
Delete a scan and all associated files.
|
|
|
|
Removes:
|
|
- Database record (cascade deletes related records)
|
|
- JSON report file
|
|
- HTML report file
|
|
- ZIP archive file
|
|
- Screenshot directory
|
|
|
|
Args:
|
|
scan_id: Scan ID to delete
|
|
|
|
Returns:
|
|
True if deleted successfully
|
|
|
|
Raises:
|
|
ValueError: If scan not found
|
|
"""
|
|
scan = self.db.query(Scan).filter(Scan.id == scan_id).first()
|
|
if not scan:
|
|
raise ValueError(f"Scan {scan_id} not found")
|
|
|
|
logger.info(f"Deleting scan {scan_id}")
|
|
|
|
# Delete files (handle missing files gracefully)
|
|
files_to_delete = [
|
|
scan.json_path,
|
|
scan.html_path,
|
|
scan.zip_path
|
|
]
|
|
|
|
for file_path in files_to_delete:
|
|
if file_path:
|
|
try:
|
|
Path(file_path).unlink()
|
|
logger.debug(f"Deleted file: {file_path}")
|
|
except FileNotFoundError:
|
|
logger.warning(f"File not found (already deleted?): {file_path}")
|
|
except Exception as e:
|
|
logger.error(f"Error deleting file {file_path}: {e}")
|
|
|
|
# Delete screenshot directory
|
|
if scan.screenshot_dir:
|
|
try:
|
|
shutil.rmtree(scan.screenshot_dir)
|
|
logger.debug(f"Deleted directory: {scan.screenshot_dir}")
|
|
except FileNotFoundError:
|
|
logger.warning(f"Directory not found (already deleted?): {scan.screenshot_dir}")
|
|
except Exception as e:
|
|
logger.error(f"Error deleting directory {scan.screenshot_dir}: {e}")
|
|
|
|
# Delete database record (cascade handles related records)
|
|
self.db.delete(scan)
|
|
self.db.commit()
|
|
|
|
logger.info(f"Scan {scan_id} deleted successfully")
|
|
return True
|
|
|
|
def get_scan_status(self, scan_id: int) -> Optional[Dict[str, Any]]:
|
|
"""
|
|
Get current scan status and progress.
|
|
|
|
Args:
|
|
scan_id: Scan ID
|
|
|
|
Returns:
|
|
Dictionary with status information, or None if scan not found
|
|
"""
|
|
scan = self.db.query(Scan).filter(Scan.id == scan_id).first()
|
|
if not scan:
|
|
return None
|
|
|
|
status_info = {
|
|
'scan_id': scan.id,
|
|
'status': scan.status,
|
|
'title': scan.title,
|
|
'timestamp': scan.timestamp.isoformat() if scan.timestamp else None,
|
|
'started_at': scan.started_at.isoformat() if scan.started_at else None,
|
|
'completed_at': scan.completed_at.isoformat() if scan.completed_at else None,
|
|
'duration': scan.duration,
|
|
'triggered_by': scan.triggered_by
|
|
}
|
|
|
|
# Add progress estimate based on status
|
|
if scan.status == 'running':
|
|
status_info['progress'] = 'In progress'
|
|
elif scan.status == 'completed':
|
|
status_info['progress'] = 'Complete'
|
|
elif scan.status == 'failed':
|
|
status_info['progress'] = 'Failed'
|
|
status_info['error_message'] = scan.error_message
|
|
|
|
return status_info
|
|
|
|
def cleanup_orphaned_scans(self) -> int:
|
|
"""
|
|
Clean up orphaned scans that are stuck in 'running' status.
|
|
|
|
This should be called on application startup to handle scans that
|
|
were running when the system crashed or was restarted.
|
|
|
|
Scans in 'running' status are marked as 'failed' with an appropriate
|
|
error message indicating they were orphaned.
|
|
|
|
Returns:
|
|
Number of orphaned scans cleaned up
|
|
"""
|
|
# Find all scans with status='running'
|
|
orphaned_scans = self.db.query(Scan).filter(Scan.status == 'running').all()
|
|
|
|
if not orphaned_scans:
|
|
logger.info("No orphaned scans found")
|
|
return 0
|
|
|
|
count = len(orphaned_scans)
|
|
logger.warning(f"Found {count} orphaned scan(s) in 'running' status, marking as failed")
|
|
|
|
# Mark each orphaned scan as failed
|
|
for scan in orphaned_scans:
|
|
scan.status = 'failed'
|
|
scan.completed_at = datetime.utcnow()
|
|
scan.error_message = (
|
|
"Scan was interrupted by system shutdown or crash. "
|
|
"The scan was running but did not complete normally."
|
|
)
|
|
|
|
# Calculate duration if we have a started_at time
|
|
if scan.started_at:
|
|
duration = (datetime.utcnow() - scan.started_at).total_seconds()
|
|
scan.duration = duration
|
|
|
|
logger.info(
|
|
f"Marked orphaned scan {scan.id} as failed "
|
|
f"(started: {scan.started_at.isoformat() if scan.started_at else 'unknown'})"
|
|
)
|
|
|
|
self.db.commit()
|
|
logger.info(f"Cleaned up {count} orphaned scan(s)")
|
|
|
|
return count
|
|
|
|
def _save_scan_to_db(self, report: Dict[str, Any], scan_id: int,
|
|
status: str = 'completed', output_paths: Dict = None) -> None:
|
|
"""
|
|
Save scan results to database.
|
|
|
|
Updates the Scan record and creates all related records
|
|
(sites, IPs, ports, services, certificates, TLS versions).
|
|
|
|
Args:
|
|
report: Scan report dictionary from scanner
|
|
scan_id: Scan ID to update
|
|
status: Final scan status (completed or failed)
|
|
output_paths: Dictionary with paths to generated files {'json': Path, 'html': Path, 'zip': Path}
|
|
"""
|
|
scan = self.db.query(Scan).filter(Scan.id == scan_id).first()
|
|
if not scan:
|
|
raise ValueError(f"Scan {scan_id} not found")
|
|
|
|
# Update scan record
|
|
scan.status = status
|
|
scan.duration = report.get('scan_duration')
|
|
scan.completed_at = datetime.utcnow()
|
|
|
|
# Save output file paths
|
|
if output_paths:
|
|
if 'json' in output_paths:
|
|
scan.json_path = str(output_paths['json'])
|
|
if 'html' in output_paths:
|
|
scan.html_path = str(output_paths['html'])
|
|
if 'zip' in output_paths:
|
|
scan.zip_path = str(output_paths['zip'])
|
|
if 'screenshots' in output_paths:
|
|
scan.screenshot_dir = str(output_paths['screenshots'])
|
|
|
|
# Map report data to database models
|
|
self._map_report_to_models(report, scan)
|
|
|
|
self.db.commit()
|
|
logger.info(f"Scan {scan_id} saved to database with status '{status}'")
|
|
|
|
def _map_report_to_models(self, report: Dict[str, Any], scan_obj: Scan) -> None:
|
|
"""
|
|
Map JSON report structure to database models.
|
|
|
|
Creates records for sites, IPs, ports, services, certificates, and TLS versions.
|
|
Processes nested relationships in order to handle foreign keys correctly.
|
|
|
|
Args:
|
|
report: Scan report dictionary
|
|
scan_obj: Scan database object to attach records to
|
|
"""
|
|
logger.debug(f"Mapping report to database models for scan {scan_obj.id}")
|
|
|
|
# Process each site
|
|
for site_data in report.get('sites', []):
|
|
# Create ScanSite record
|
|
site = ScanSite(
|
|
scan_id=scan_obj.id,
|
|
site_name=site_data['name']
|
|
)
|
|
self.db.add(site)
|
|
self.db.flush() # Get site.id for foreign key
|
|
|
|
# Create ScanSiteAssociation if this site exists in the database
|
|
# This links the scan to reusable site definitions
|
|
master_site = (
|
|
self.db.query(Site)
|
|
.filter(Site.name == site_data['name'])
|
|
.first()
|
|
)
|
|
|
|
if master_site:
|
|
# Check if association already exists (avoid duplicates)
|
|
existing_assoc = (
|
|
self.db.query(ScanSiteAssociation)
|
|
.filter(
|
|
ScanSiteAssociation.scan_id == scan_obj.id,
|
|
ScanSiteAssociation.site_id == master_site.id
|
|
)
|
|
.first()
|
|
)
|
|
|
|
if not existing_assoc:
|
|
assoc = ScanSiteAssociation(
|
|
scan_id=scan_obj.id,
|
|
site_id=master_site.id,
|
|
created_at=datetime.utcnow()
|
|
)
|
|
self.db.add(assoc)
|
|
logger.debug(f"Created association between scan {scan_obj.id} and site '{master_site.name}' (id={master_site.id})")
|
|
|
|
# Process each IP in this site
|
|
for ip_data in site_data.get('ips', []):
|
|
# Create ScanIP record
|
|
ip = ScanIP(
|
|
scan_id=scan_obj.id,
|
|
site_id=site.id,
|
|
ip_address=ip_data['address'],
|
|
ping_expected=ip_data.get('expected', {}).get('ping'),
|
|
ping_actual=ip_data.get('actual', {}).get('ping')
|
|
)
|
|
self.db.add(ip)
|
|
self.db.flush()
|
|
|
|
# Process TCP ports
|
|
expected_tcp = set(ip_data.get('expected', {}).get('tcp_ports', []))
|
|
actual_tcp = ip_data.get('actual', {}).get('tcp_ports', [])
|
|
|
|
for port_num in actual_tcp:
|
|
port = ScanPort(
|
|
scan_id=scan_obj.id,
|
|
ip_id=ip.id,
|
|
port=port_num,
|
|
protocol='tcp',
|
|
expected=(port_num in expected_tcp),
|
|
state='open'
|
|
)
|
|
self.db.add(port)
|
|
self.db.flush()
|
|
|
|
# Find service for this port
|
|
service_data = self._find_service_for_port(
|
|
ip_data.get('actual', {}).get('services', []),
|
|
port_num
|
|
)
|
|
|
|
if service_data:
|
|
# Create ScanService record
|
|
service = ScanServiceModel(
|
|
scan_id=scan_obj.id,
|
|
port_id=port.id,
|
|
service_name=service_data.get('service'),
|
|
product=service_data.get('product'),
|
|
version=service_data.get('version'),
|
|
extrainfo=service_data.get('extrainfo'),
|
|
ostype=service_data.get('ostype'),
|
|
http_protocol=service_data.get('http_info', {}).get('protocol'),
|
|
screenshot_path=service_data.get('http_info', {}).get('screenshot')
|
|
)
|
|
self.db.add(service)
|
|
self.db.flush()
|
|
|
|
# Process certificate and TLS info if present
|
|
http_info = service_data.get('http_info', {})
|
|
ssl_tls = http_info.get('ssl_tls', {})
|
|
if ssl_tls.get('certificate'):
|
|
self._process_certificate(
|
|
ssl_tls,
|
|
scan_obj.id,
|
|
service.id
|
|
)
|
|
|
|
# Process UDP ports
|
|
expected_udp = set(ip_data.get('expected', {}).get('udp_ports', []))
|
|
actual_udp = ip_data.get('actual', {}).get('udp_ports', [])
|
|
|
|
for port_num in actual_udp:
|
|
port = ScanPort(
|
|
scan_id=scan_obj.id,
|
|
ip_id=ip.id,
|
|
port=port_num,
|
|
protocol='udp',
|
|
expected=(port_num in expected_udp),
|
|
state='open'
|
|
)
|
|
self.db.add(port)
|
|
|
|
logger.debug(f"Report mapping complete for scan {scan_obj.id}")
|
|
|
|
def _find_service_for_port(self, services: List[Dict], port: int) -> Optional[Dict]:
|
|
"""
|
|
Find service data for a specific port.
|
|
|
|
Args:
|
|
services: List of service dictionaries
|
|
port: Port number to find
|
|
|
|
Returns:
|
|
Service dictionary if found, None otherwise
|
|
"""
|
|
for service in services:
|
|
if service.get('port') == port:
|
|
return service
|
|
return None
|
|
|
|
def _process_certificate(self, ssl_tls_data: Dict[str, Any], scan_id: int,
|
|
service_id: int) -> None:
|
|
"""
|
|
Process certificate and TLS version data.
|
|
|
|
Args:
|
|
ssl_tls_data: SSL/TLS data dictionary containing 'certificate' and 'tls_versions'
|
|
scan_id: Scan ID
|
|
service_id: Service ID
|
|
"""
|
|
# Extract certificate data from ssl_tls structure
|
|
cert_data = ssl_tls_data.get('certificate', {})
|
|
|
|
# Create ScanCertificate record
|
|
cert = ScanCertificate(
|
|
scan_id=scan_id,
|
|
service_id=service_id,
|
|
subject=cert_data.get('subject'),
|
|
issuer=cert_data.get('issuer'),
|
|
serial_number=cert_data.get('serial_number'),
|
|
not_valid_before=self._parse_datetime(cert_data.get('not_valid_before')),
|
|
not_valid_after=self._parse_datetime(cert_data.get('not_valid_after')),
|
|
days_until_expiry=cert_data.get('days_until_expiry'),
|
|
sans=json.dumps(cert_data.get('sans', [])),
|
|
is_self_signed=cert_data.get('is_self_signed', False)
|
|
)
|
|
self.db.add(cert)
|
|
self.db.flush()
|
|
|
|
# Process TLS versions
|
|
tls_versions = ssl_tls_data.get('tls_versions', {})
|
|
for version, version_data in tls_versions.items():
|
|
tls = ScanTLSVersion(
|
|
scan_id=scan_id,
|
|
certificate_id=cert.id,
|
|
tls_version=version,
|
|
supported=version_data.get('supported', False),
|
|
cipher_suites=json.dumps(version_data.get('cipher_suites', []))
|
|
)
|
|
self.db.add(tls)
|
|
|
|
def _parse_datetime(self, date_str: Optional[str]) -> Optional[datetime]:
|
|
"""
|
|
Parse ISO datetime string.
|
|
|
|
Args:
|
|
date_str: ISO format datetime string
|
|
|
|
Returns:
|
|
datetime object or None if parsing fails
|
|
"""
|
|
if not date_str:
|
|
return None
|
|
|
|
try:
|
|
# Handle ISO format with 'Z' suffix
|
|
if date_str.endswith('Z'):
|
|
date_str = date_str[:-1] + '+00:00'
|
|
return datetime.fromisoformat(date_str)
|
|
except (ValueError, AttributeError) as e:
|
|
logger.warning(f"Failed to parse datetime '{date_str}': {e}")
|
|
return None
|
|
|
|
def _scan_to_dict(self, scan: Scan) -> Dict[str, Any]:
|
|
"""
|
|
Convert Scan object to dictionary with full details.
|
|
|
|
Args:
|
|
scan: Scan database object
|
|
|
|
Returns:
|
|
Dictionary representation with all related data
|
|
"""
|
|
return {
|
|
'id': scan.id,
|
|
'timestamp': scan.timestamp.isoformat() if scan.timestamp else None,
|
|
'duration': scan.duration,
|
|
'status': scan.status,
|
|
'title': scan.title,
|
|
'config_id': scan.config_id,
|
|
'json_path': scan.json_path,
|
|
'html_path': scan.html_path,
|
|
'zip_path': scan.zip_path,
|
|
'screenshot_dir': scan.screenshot_dir,
|
|
'triggered_by': scan.triggered_by,
|
|
'created_at': scan.created_at.isoformat() if scan.created_at else None,
|
|
'sites': [self._site_to_dict(site) for site in scan.sites]
|
|
}
|
|
|
|
def _scan_to_summary_dict(self, scan: Scan) -> Dict[str, Any]:
|
|
"""
|
|
Convert Scan object to summary dictionary (no related data).
|
|
|
|
Args:
|
|
scan: Scan database object
|
|
|
|
Returns:
|
|
Summary dictionary
|
|
"""
|
|
return {
|
|
'id': scan.id,
|
|
'timestamp': scan.timestamp.isoformat() if scan.timestamp else None,
|
|
'duration': scan.duration,
|
|
'status': scan.status,
|
|
'title': scan.title,
|
|
'config_id': scan.config_id,
|
|
'triggered_by': scan.triggered_by,
|
|
'created_at': scan.created_at.isoformat() if scan.created_at else None
|
|
}
|
|
|
|
def _site_to_dict(self, site: ScanSite) -> Dict[str, Any]:
|
|
"""Convert ScanSite to dictionary."""
|
|
return {
|
|
'id': site.id,
|
|
'name': site.site_name,
|
|
'ips': [self._ip_to_dict(ip) for ip in site.ips]
|
|
}
|
|
|
|
def _ip_to_dict(self, ip: ScanIP) -> Dict[str, Any]:
|
|
"""Convert ScanIP to dictionary."""
|
|
return {
|
|
'id': ip.id,
|
|
'address': ip.ip_address,
|
|
'ping_expected': ip.ping_expected,
|
|
'ping_actual': ip.ping_actual,
|
|
'ports': [self._port_to_dict(port) for port in ip.ports]
|
|
}
|
|
|
|
def _port_to_dict(self, port: ScanPort) -> Dict[str, Any]:
|
|
"""Convert ScanPort to dictionary."""
|
|
return {
|
|
'id': port.id,
|
|
'port': port.port,
|
|
'protocol': port.protocol,
|
|
'state': port.state,
|
|
'expected': port.expected,
|
|
'services': [self._service_to_dict(svc) for svc in port.services]
|
|
}
|
|
|
|
def _service_to_dict(self, service: ScanServiceModel) -> Dict[str, Any]:
|
|
"""Convert ScanService to dictionary."""
|
|
result = {
|
|
'id': service.id,
|
|
'service_name': service.service_name,
|
|
'product': service.product,
|
|
'version': service.version,
|
|
'extrainfo': service.extrainfo,
|
|
'ostype': service.ostype,
|
|
'http_protocol': service.http_protocol,
|
|
'screenshot_path': service.screenshot_path
|
|
}
|
|
|
|
# Add certificate info if present
|
|
if service.certificates:
|
|
result['certificates'] = [
|
|
self._certificate_to_dict(cert) for cert in service.certificates
|
|
]
|
|
|
|
return result
|
|
|
|
def _certificate_to_dict(self, cert: ScanCertificate) -> Dict[str, Any]:
|
|
"""Convert ScanCertificate to dictionary."""
|
|
result = {
|
|
'id': cert.id,
|
|
'subject': cert.subject,
|
|
'issuer': cert.issuer,
|
|
'serial_number': cert.serial_number,
|
|
'not_valid_before': cert.not_valid_before.isoformat() if cert.not_valid_before else None,
|
|
'not_valid_after': cert.not_valid_after.isoformat() if cert.not_valid_after else None,
|
|
'days_until_expiry': cert.days_until_expiry,
|
|
'is_self_signed': cert.is_self_signed
|
|
}
|
|
|
|
# Parse SANs from JSON
|
|
if cert.sans:
|
|
try:
|
|
result['sans'] = json.loads(cert.sans)
|
|
except json.JSONDecodeError:
|
|
result['sans'] = []
|
|
|
|
# Add TLS versions
|
|
result['tls_versions'] = [
|
|
self._tls_version_to_dict(tls) for tls in cert.tls_versions
|
|
]
|
|
|
|
return result
|
|
|
|
def _tls_version_to_dict(self, tls: ScanTLSVersion) -> Dict[str, Any]:
|
|
"""Convert ScanTLSVersion to dictionary."""
|
|
result = {
|
|
'id': tls.id,
|
|
'tls_version': tls.tls_version,
|
|
'supported': tls.supported
|
|
}
|
|
|
|
# Parse cipher suites from JSON
|
|
if tls.cipher_suites:
|
|
try:
|
|
result['cipher_suites'] = json.loads(tls.cipher_suites)
|
|
except json.JSONDecodeError:
|
|
result['cipher_suites'] = []
|
|
|
|
return result
|
|
|
|
def compare_scans(self, scan1_id: int, scan2_id: int) -> Optional[Dict[str, Any]]:
|
|
"""
|
|
Compare two scans and return the differences.
|
|
|
|
Compares ports, services, and certificates between two scans,
|
|
highlighting added, removed, and changed items.
|
|
|
|
Args:
|
|
scan1_id: ID of the first (older) scan
|
|
scan2_id: ID of the second (newer) scan
|
|
|
|
Returns:
|
|
Dictionary with comparison results, or None if either scan not found
|
|
{
|
|
'scan1': {...}, # Scan 1 summary
|
|
'scan2': {...}, # Scan 2 summary
|
|
'same_config': bool, # Whether both scans used the same config
|
|
'config_warning': str | None, # Warning message if configs differ
|
|
'ports': {
|
|
'added': [...],
|
|
'removed': [...],
|
|
'unchanged': [...]
|
|
},
|
|
'services': {
|
|
'added': [...],
|
|
'removed': [...],
|
|
'changed': [...]
|
|
},
|
|
'certificates': {
|
|
'added': [...],
|
|
'removed': [...],
|
|
'changed': [...]
|
|
},
|
|
'drift_score': 0.0-1.0
|
|
}
|
|
"""
|
|
# Get both scans
|
|
scan1 = self.get_scan(scan1_id)
|
|
scan2 = self.get_scan(scan2_id)
|
|
|
|
if not scan1 or not scan2:
|
|
return None
|
|
|
|
# Check if scans use the same configuration
|
|
config1 = scan1.get('config_id')
|
|
config2 = scan2.get('config_id')
|
|
same_config = (config1 == config2) and (config1 is not None)
|
|
|
|
# Generate warning message if configs differ
|
|
config_warning = None
|
|
if not same_config:
|
|
config_warning = (
|
|
f"These scans use different configurations. "
|
|
f"Scan #{scan1_id} used config_id={config1 or 'unknown'} and "
|
|
f"Scan #{scan2_id} used config_id={config2 or 'unknown'}. "
|
|
f"The comparison may show all changes as additions/removals if the scans "
|
|
f"cover different IP ranges or infrastructure."
|
|
)
|
|
|
|
# Extract port data
|
|
ports1 = self._extract_ports_from_scan(scan1)
|
|
ports2 = self._extract_ports_from_scan(scan2)
|
|
|
|
# Compare ports
|
|
ports_comparison = self._compare_ports(ports1, ports2)
|
|
|
|
# Extract service data
|
|
services1 = self._extract_services_from_scan(scan1)
|
|
services2 = self._extract_services_from_scan(scan2)
|
|
|
|
# Compare services
|
|
services_comparison = self._compare_services(services1, services2)
|
|
|
|
# Extract certificate data
|
|
certs1 = self._extract_certificates_from_scan(scan1)
|
|
certs2 = self._extract_certificates_from_scan(scan2)
|
|
|
|
# Compare certificates
|
|
certificates_comparison = self._compare_certificates(certs1, certs2)
|
|
|
|
# Calculate drift score (0.0 = identical, 1.0 = completely different)
|
|
drift_score = self._calculate_drift_score(
|
|
ports_comparison,
|
|
services_comparison,
|
|
certificates_comparison
|
|
)
|
|
|
|
return {
|
|
'scan1': {
|
|
'id': scan1['id'],
|
|
'timestamp': scan1['timestamp'],
|
|
'title': scan1['title'],
|
|
'status': scan1['status'],
|
|
'config_id': config1
|
|
},
|
|
'scan2': {
|
|
'id': scan2['id'],
|
|
'timestamp': scan2['timestamp'],
|
|
'title': scan2['title'],
|
|
'status': scan2['status'],
|
|
'config_id': config2
|
|
},
|
|
'same_config': same_config,
|
|
'config_warning': config_warning,
|
|
'ports': ports_comparison,
|
|
'services': services_comparison,
|
|
'certificates': certificates_comparison,
|
|
'drift_score': drift_score
|
|
}
|
|
|
|
def _extract_ports_from_scan(self, scan: Dict[str, Any]) -> Dict[str, Any]:
|
|
"""
|
|
Extract port information from a scan.
|
|
|
|
Returns:
|
|
Dictionary mapping "ip:port:protocol" to port details
|
|
"""
|
|
ports = {}
|
|
for site in scan.get('sites', []):
|
|
for ip_data in site.get('ips', []):
|
|
ip_addr = ip_data['address']
|
|
for port_data in ip_data.get('ports', []):
|
|
key = f"{ip_addr}:{port_data['port']}:{port_data['protocol']}"
|
|
ports[key] = {
|
|
'ip': ip_addr,
|
|
'port': port_data['port'],
|
|
'protocol': port_data['protocol'],
|
|
'state': port_data.get('state', 'unknown'),
|
|
'expected': port_data.get('expected')
|
|
}
|
|
return ports
|
|
|
|
def _extract_services_from_scan(self, scan: Dict[str, Any]) -> Dict[str, Any]:
|
|
"""
|
|
Extract service information from a scan.
|
|
|
|
Returns:
|
|
Dictionary mapping "ip:port:protocol" to service details
|
|
"""
|
|
services = {}
|
|
for site in scan.get('sites', []):
|
|
for ip_data in site.get('ips', []):
|
|
ip_addr = ip_data['address']
|
|
for port_data in ip_data.get('ports', []):
|
|
port_num = port_data['port']
|
|
protocol = port_data['protocol']
|
|
key = f"{ip_addr}:{port_num}:{protocol}"
|
|
|
|
# Get first service (usually only one per port)
|
|
port_services = port_data.get('services', [])
|
|
if port_services:
|
|
svc = port_services[0]
|
|
services[key] = {
|
|
'ip': ip_addr,
|
|
'port': port_num,
|
|
'protocol': protocol,
|
|
'service_name': svc.get('service_name'),
|
|
'product': svc.get('product'),
|
|
'version': svc.get('version'),
|
|
'extrainfo': svc.get('extrainfo')
|
|
}
|
|
return services
|
|
|
|
def _extract_certificates_from_scan(self, scan: Dict[str, Any]) -> Dict[str, Any]:
|
|
"""
|
|
Extract certificate information from a scan.
|
|
|
|
Returns:
|
|
Dictionary mapping "ip:port" to certificate details
|
|
"""
|
|
certificates = {}
|
|
for site in scan.get('sites', []):
|
|
for ip_data in site.get('ips', []):
|
|
ip_addr = ip_data['address']
|
|
for port_data in ip_data.get('ports', []):
|
|
port_num = port_data['port']
|
|
protocol = port_data['protocol']
|
|
|
|
# Get certificates from services
|
|
for svc in port_data.get('services', []):
|
|
if svc.get('certificates'):
|
|
for cert in svc['certificates']:
|
|
key = f"{ip_addr}:{port_num}"
|
|
certificates[key] = {
|
|
'ip': ip_addr,
|
|
'port': port_num,
|
|
'subject': cert.get('subject'),
|
|
'issuer': cert.get('issuer'),
|
|
'not_valid_after': cert.get('not_valid_after'),
|
|
'days_until_expiry': cert.get('days_until_expiry'),
|
|
'is_self_signed': cert.get('is_self_signed')
|
|
}
|
|
return certificates
|
|
|
|
def _compare_ports(self, ports1: Dict, ports2: Dict) -> Dict[str, List]:
|
|
"""
|
|
Compare port sets between two scans.
|
|
|
|
Returns:
|
|
Dictionary with added, removed, and unchanged ports
|
|
"""
|
|
keys1 = set(ports1.keys())
|
|
keys2 = set(ports2.keys())
|
|
|
|
added_keys = keys2 - keys1
|
|
removed_keys = keys1 - keys2
|
|
unchanged_keys = keys1 & keys2
|
|
|
|
return {
|
|
'added': [ports2[k] for k in sorted(added_keys)],
|
|
'removed': [ports1[k] for k in sorted(removed_keys)],
|
|
'unchanged': [ports2[k] for k in sorted(unchanged_keys)]
|
|
}
|
|
|
|
def _compare_services(self, services1: Dict, services2: Dict) -> Dict[str, List]:
|
|
"""
|
|
Compare services between two scans.
|
|
|
|
Returns:
|
|
Dictionary with added, removed, and changed services
|
|
"""
|
|
keys1 = set(services1.keys())
|
|
keys2 = set(services2.keys())
|
|
|
|
added_keys = keys2 - keys1
|
|
removed_keys = keys1 - keys2
|
|
common_keys = keys1 & keys2
|
|
|
|
# Find changed services (same port, different version/product)
|
|
changed = []
|
|
for key in sorted(common_keys):
|
|
svc1 = services1[key]
|
|
svc2 = services2[key]
|
|
|
|
# Check if service details changed
|
|
if (svc1.get('product') != svc2.get('product') or
|
|
svc1.get('version') != svc2.get('version') or
|
|
svc1.get('service_name') != svc2.get('service_name')):
|
|
changed.append({
|
|
'ip': svc2['ip'],
|
|
'port': svc2['port'],
|
|
'protocol': svc2['protocol'],
|
|
'old': {
|
|
'service_name': svc1.get('service_name'),
|
|
'product': svc1.get('product'),
|
|
'version': svc1.get('version')
|
|
},
|
|
'new': {
|
|
'service_name': svc2.get('service_name'),
|
|
'product': svc2.get('product'),
|
|
'version': svc2.get('version')
|
|
}
|
|
})
|
|
|
|
return {
|
|
'added': [services2[k] for k in sorted(added_keys)],
|
|
'removed': [services1[k] for k in sorted(removed_keys)],
|
|
'changed': changed
|
|
}
|
|
|
|
def _compare_certificates(self, certs1: Dict, certs2: Dict) -> Dict[str, List]:
|
|
"""
|
|
Compare certificates between two scans.
|
|
|
|
Returns:
|
|
Dictionary with added, removed, and changed certificates
|
|
"""
|
|
keys1 = set(certs1.keys())
|
|
keys2 = set(certs2.keys())
|
|
|
|
added_keys = keys2 - keys1
|
|
removed_keys = keys1 - keys2
|
|
common_keys = keys1 & keys2
|
|
|
|
# Find changed certificates (same IP:port, different cert details)
|
|
changed = []
|
|
for key in sorted(common_keys):
|
|
cert1 = certs1[key]
|
|
cert2 = certs2[key]
|
|
|
|
# Check if certificate changed
|
|
if (cert1.get('subject') != cert2.get('subject') or
|
|
cert1.get('issuer') != cert2.get('issuer') or
|
|
cert1.get('not_valid_after') != cert2.get('not_valid_after')):
|
|
changed.append({
|
|
'ip': cert2['ip'],
|
|
'port': cert2['port'],
|
|
'old': {
|
|
'subject': cert1.get('subject'),
|
|
'issuer': cert1.get('issuer'),
|
|
'not_valid_after': cert1.get('not_valid_after'),
|
|
'days_until_expiry': cert1.get('days_until_expiry')
|
|
},
|
|
'new': {
|
|
'subject': cert2.get('subject'),
|
|
'issuer': cert2.get('issuer'),
|
|
'not_valid_after': cert2.get('not_valid_after'),
|
|
'days_until_expiry': cert2.get('days_until_expiry')
|
|
}
|
|
})
|
|
|
|
return {
|
|
'added': [certs2[k] for k in sorted(added_keys)],
|
|
'removed': [certs1[k] for k in sorted(removed_keys)],
|
|
'changed': changed
|
|
}
|
|
|
|
def _calculate_drift_score(self, ports_comp: Dict, services_comp: Dict,
|
|
certs_comp: Dict) -> float:
|
|
"""
|
|
Calculate drift score based on comparison results.
|
|
|
|
Returns:
|
|
Float between 0.0 (identical) and 1.0 (completely different)
|
|
"""
|
|
# Count total items in both scans
|
|
total_ports = (
|
|
len(ports_comp['added']) +
|
|
len(ports_comp['removed']) +
|
|
len(ports_comp['unchanged'])
|
|
)
|
|
|
|
total_services = (
|
|
len(services_comp['added']) +
|
|
len(services_comp['removed']) +
|
|
len(services_comp['changed']) +
|
|
max(0, len(ports_comp['unchanged']) - len(services_comp['changed']))
|
|
)
|
|
|
|
# Count changed items
|
|
changed_ports = len(ports_comp['added']) + len(ports_comp['removed'])
|
|
changed_services = (
|
|
len(services_comp['added']) +
|
|
len(services_comp['removed']) +
|
|
len(services_comp['changed'])
|
|
)
|
|
changed_certs = (
|
|
len(certs_comp['added']) +
|
|
len(certs_comp['removed']) +
|
|
len(certs_comp['changed'])
|
|
)
|
|
|
|
# Calculate weighted drift score
|
|
# Ports have 50% weight, services 30%, certificates 20%
|
|
port_drift = changed_ports / max(total_ports, 1)
|
|
service_drift = changed_services / max(total_services, 1)
|
|
cert_drift = changed_certs / max(len(certs_comp['added']) + len(certs_comp['removed']) + len(certs_comp['changed']), 1)
|
|
|
|
drift_score = (port_drift * 0.5) + (service_drift * 0.3) + (cert_drift * 0.2)
|
|
|
|
return round(min(drift_score, 1.0), 3) # Cap at 1.0 and round to 3 decimals
|