Implementing Cross-Registry Scanning Orchestration

Create a unified scanning solution across multiple registries:

#!/usr/bin/env python3
# unified-registry-scanner.py

import asyncio
import aiohttp
from typing import List, Dict, Any
import json
from abc import ABC, abstractmethod

class RegistryScanner(ABC):
    @abstractmethod
    async def scan_image(self, image: str) -> Dict[str, Any]:
        pass
    
    @abstractmethod
    async def get_repositories(self) -> List[str]:
        pass

class TrivyScanner:
    async def scan(self, image: str) -> Dict[str, Any]:
        async with aiohttp.ClientSession() as session:
            # Assuming Trivy server is running
            async with session.post(
                'http://trivy-server:8080/scan',
                json={'image': image}
            ) as response:
                return await response.json()

class UnifiedRegistryScanner:
    def __init__(self):
        self.scanners = []
        self.registries = []
        self.trivy = TrivyScanner()
        
    def add_registry(self, registry_type: str, config: Dict[str, Any]):
        if registry_type == 'harbor':
            self.registries.append(HarborRegistry(config))
        elif registry_type == 'ecr':
            self.registries.append(ECRRegistry(config))
        elif registry_type == 'gcr':
            self.registries.append(GCRRegistry(config))
        elif registry_type == 'dockerhub':
            self.registries.append(DockerHubRegistry(config))
            
    async def scan_all_registries(self) -> Dict[str, Any]:
        """Scan all configured registries"""
        results = {
            'scan_timestamp': datetime.now().isoformat(),
            'registries': {},
            'summary': {
                'total_images': 0,
                'vulnerable_images': 0,
                'critical_findings': 0,
                'high_findings': 0
            }
        }
        
        # Scan each registry concurrently
        tasks = []
        for registry in self.registries:
            tasks.append(self.scan_registry(registry))
            
        registry_results = await asyncio.gather(*tasks)
        
        # Aggregate results
        for registry, result in zip(self.registries, registry_results):
            results['registries'][registry.name] = result
            results['summary']['total_images'] += result['total_images']
            results['summary']['vulnerable_images'] += result['vulnerable_images']
            results['summary']['critical_findings'] += result['critical_findings']
            results['summary']['high_findings'] += result['high_findings']
            
        return results
        
    async def scan_registry(self, registry: RegistryScanner) -> Dict[str, Any]:
        """Scan a single registry"""
        result = {
            'name': registry.name,
            'type': registry.type,
            'total_images': 0,
            'vulnerable_images': 0,
            'critical_findings': 0,
            'high_findings': 0,
            'images': []
        }
        
        # Get all repositories
        repositories = await registry.get_repositories()
        
        # Scan each image
        scan_tasks = []
        for repo in repositories:
            images = await registry.get_images(repo)
            for image in images:
                result['total_images'] += 1
                scan_tasks.append(self.scan_image(registry, image))
                
        # Run scans concurrently with rate limiting
        semaphore = asyncio.Semaphore(10)  # Max 10 concurrent scans
        
        async def scan_with_limit(image_scan):
            async with semaphore:
                return await image_scan
                
        scan_results = await asyncio.gather(
            *[scan_with_limit(task) for task in scan_tasks]
        )
        
        # Process results
        for scan_result in scan_results:
            if scan_result['vulnerabilities']:
                result['vulnerable_images'] += 1
                
                for vuln in scan_result['vulnerabilities']:
                    if vuln['severity'] == 'CRITICAL':
                        result['critical_findings'] += 1
                    elif vuln['severity'] == 'HIGH':
                        result['high_findings'] += 1
                        
                result['images'].append({
                    'name': scan_result['image'],
                    'critical': scan_result['summary']['critical'],
                    'high': scan_result['summary']['high'],
                    'scan_time': scan_result['scan_time']
                })
                
        return result

# Orchestration configuration
config = {
    'registries': [
        {
            'type': 'harbor',
            'name': 'production-harbor',
            'url': 'https://harbor.company.com',
            'credentials': {'username': 'scanner', 'password': 'secret'}
        },
        {
            'type': 'ecr',
            'name': 'aws-ecr',
            'region': 'us-east-1',
            'credentials': {'profile': 'production'}
        },
        {
            'type': 'gcr',
            'name': 'gcp-gcr',
            'project': 'my-project',
            'credentials': {'key_file': '/path/to/key.json'}
        }
    ],
    'notifications': {
        'slack': {
            'webhook': 'https://hooks.slack.com/...',
            'channel': '#security-alerts'
        },
        'email': {
            'smtp_server': 'smtp.company.com',
            'recipients': ['[email protected]']
        }
    }
}

# Main execution
async def main():
    scanner = UnifiedRegistryScanner()
    
    # Configure registries
    for reg_config in config['registries']:
        scanner.add_registry(reg_config['type'], reg_config)
        
    # Run scans
    results = await scanner.scan_all_registries()
    
    # Save results
    with open('unified-scan-results.json', 'w') as f:
        json.dump(results, f, indent=2)
        
    # Send notifications if critical findings
    if results['summary']['critical_findings'] > 0:
        await send_notifications(results, config['notifications'])
        
if __name__ == '__main__':
    asyncio.run(main())

Container registry scanning automation provides continuous security visibility for your entire image inventory. By implementing comprehensive scanning across different registry platforms and creating unified monitoring solutions, organizations can maintain security at scale. The next chapter explores detecting and responding to CVEs in container environments.## CVE Detection in Container Images

Common Vulnerabilities and Exposures (CVEs) represent the standardized method for identifying and cataloging security vulnerabilities across software systems. In containerized environments, CVE detection becomes particularly complex due to the layered nature of container images and the diverse sources of vulnerabilities they can contain. This chapter provides comprehensive guidance on detecting, understanding, and responding to CVEs in container images, using both Trivy and Snyk to implement robust CVE management strategies.