Automating Vulnerability Checks with Python
In previous articles, we’ve explored API automation, file parsing, and system monitoring. Now we’re taking those skills into security territory—specifically, automating vulnerability checks. This isn’t about reinventing Nessus or OpenVAS; it’s about building custom tooling that fits your infrastructure, automates repetitive security tasks, and integrates with your existing workflows.
Security teams are drowning in manual checks: verifying package versions, cross-referencing CVE databases, scanning configuration files, and tracking remediation. Python gives us the power to automate these processes, create custom scanners tailored to our environment, and build early-warning systems that catch vulnerabilities before they’re exploited.
Table of Contents
- Why Automate Vulnerability Checks
- Integrating with CVE Databases
- Automated Dependency Scanning
- Configuration Vulnerability Checks
- Reporting and Alerting
- Production Considerations
Why Automate Vulnerability Checks
Manual vulnerability management doesn’t scale. When you’re managing dozens of servers, hundreds of containers, or thousands of dependencies, manual checks become impossible. Automation provides consistency, speed, and the ability to run checks continuously rather than quarterly.
Python excels here because it integrates easily with existing security tools, parses virtually any data format, and has robust libraries for network operations, data analysis, and API integration. You’re not replacing your security stack—you’re making it smarter and more responsive.
Integrating with CVE Databases
The National Vulnerability Database (NVD) provides a JSON API for CVE lookups. Let’s build a function that queries this API, caches results locally, and returns structured vulnerability data. This is the foundation for more complex scanning tools.
For those looking to deepen their security automation knowledge, Coursera offers comprehensive cybersecurity and Python programming tracks that pair well with hands-on projects like this.
import requests
import json
import time
from datetime import datetime, timedelta
from pathlib import Path
class CVEChecker:
def __init__(self, cache_dir='cve_cache'):
self.base_url = 'https://services.nvd.nist.gov/rest/json/cves/2.0'
self.cache_dir = Path(cache_dir)
self.cache_dir.mkdir(exist_ok=True)
self.rate_limit_delay = 6 # NVD requires 6 seconds between requests without API key
def get_cve_details(self, cve_id):
"""Fetch CVE details with local caching"""
cache_file = self.cache_dir / f"{cve_id}.json"
# Check cache first
if cache_file.exists():
cache_age = datetime.now() - datetime.fromtimestamp(cache_file.stat().st_mtime)
if cache_age < timedelta(days=7):
with open(cache_file, 'r') as f:
return json.load(f)
# Fetch from API
try:
time.sleep(self.rate_limit_delay)
response = requests.get(f'{self.base_url}?cveId={cve_id}', timeout=10)
response.raise_for_status()
data = response.json()
# Cache the result
with open(cache_file, 'w') as f:
json.dump(data, f, indent=2)
return data
except requests.exceptions.RequestException as e:
print(f"Error fetching {cve_id}: {e}")
return None
def parse_cve_severity(self, cve_data):
"""Extract CVSS score and severity"""
if not cve_data or 'vulnerabilities' not in cve_data:
return None
try:
vuln = cve_data['vulnerabilities'][0]['cve']
metrics = vuln.get('metrics', {})
# Try CVSS v3.1 first, fall back to v2
if 'cvssMetricV31' in metrics:
cvss = metrics['cvssMetricV31'][0]['cvssData']
return {
'score': cvss['baseScore'],
'severity': cvss['baseSeverity'],
'vector': cvss['vectorString']
}
elif 'cvssMetricV2' in metrics:
cvss = metrics['cvssMetricV2'][0]['cvssData']
return {
'score': cvss['baseScore'],
'severity': cvss.get('baseSeverity', 'UNKNOWN'),
'vector': cvss['vectorString']
}
except (KeyError, IndexError) as e:
print(f"Error parsing CVE data: {e}")
return None
# Example usage
checker = CVEChecker()
cve_data = checker.get_cve_details('CVE-2023-44487')
if cve_data:
severity = checker.parse_cve_severity(cve_data)
if severity:
print(f"Score: {severity['score']}, Severity: {severity['severity']}")
This class handles rate limiting, caching, and error handling—all critical for production use. The cache prevents hammering the NVD API and speeds up repeated checks dramatically.
Automated Dependency Scanning
One of the most common vulnerability vectors is outdated dependencies. Let's build a scanner that reads Python requirements files, checks versions against known vulnerabilities, and generates actionable reports.
If you're building your skills in data manipulation and API integration, platforms like DataCamp offer interactive exercises that complement real-world automation projects perfectly.
import subprocess
import json
from packaging import version as pkg_version
class DependencyScanner:
def __init__(self):
self.vulnerable_packages = []
def get_installed_packages(self):
"""Get all installed packages with versions"""
result = subprocess.run(
['pip', 'list', '--format=json'],
capture_output=True,
text=True
)
return json.loads(result.stdout)
def check_package_vulnerability(self, package_name, package_version):
"""Check a package against the PyPI safety database"""
# Using PyPI's vulnerability API
try:
url = f'https://pypi.org/pypi/{package_name}/json'
response = requests.get(url, timeout=5)
if response.status_code != 200:
return None
data = response.json()
vulnerabilities = data.get('vulnerabilities', [])
if not vulnerabilities:
return None
# Check if current version is affected
affected = []
for vuln in vulnerabilities:
vuln_ranges = vuln.get('vulnerable_versions', [])
for vuln_range in vuln_ranges:
if self.version_matches_range(package_version, vuln_range):
affected.append({
'id': vuln.get('id'),
'summary': vuln.get('summary'),
'fixed_in': vuln.get('fixed_in', [])
})
return affected if affected else None
except Exception as e:
print(f"Error checking {package_name}: {e}")
return None
def version_matches_range(self, current_version, vuln_range):
"""Simple version range checking"""
# This is simplified - production should use packaging.specifiers
try:
current = pkg_version.parse(current_version)
# Handle common patterns like "<2.0.0", ">=1.0,<1.5"
if vuln_range.startswith('<'):
max_ver = pkg_version.parse(vuln_range[1:])
return current < max_ver
return True # Conservative: assume vulnerable if unsure
except:
return True
def scan_environment(self):
"""Scan entire Python environment for vulnerabilities"""
packages = self.get_installed_packages()
results = {
'total_packages': len(packages),
'vulnerable_packages': [],
'scan_time': datetime.now().isoformat()
}
for pkg in packages:
name = pkg['name']
ver = pkg['version']
print(f"Checking {name} {ver}...")
vulns = self.check_package_vulnerability(name, ver)
if vulns:
results['vulnerable_packages'].append({
'package': name,
'version': ver,
'vulnerabilities': vulns
})
return results
def generate_report(self, results, output_file='vuln_report.json'):
"""Generate JSON report of findings"""
with open(output_file, 'w') as f:
json.dump(results, f, indent=2)
# Print summary
vuln_count = len(results['vulnerable_packages'])
print(f"\n{'='*60}")
print(f"Scan complete: {vuln_count} vulnerable packages found")
print(f"Report saved to {output_file}")
if vuln_count > 0:
print("\nVulnerable packages:")
for pkg in results['vulnerable_packages']:
print(f" - {pkg['package']} {pkg['version']}: {len(pkg['vulnerabilities'])} CVEs")
# Run the scan
scanner = DependencyScanner()
scan_results = scanner.scan_environment()
scanner.generate_report(scan_results)
Configuration Vulnerability Checks
Beyond dependencies, misconfigurations are a major vulnerability source. Let's create a checker for common security misconfigurations in web server configs, SSH settings, and firewall rules.
SSH Configuration Auditing
This example scans SSH daemon configurations for common security weaknesses:
class SSHConfigChecker:
def __init__(self, config_path='/etc/ssh/sshd_config'):
self.config_path = config_path
self.findings = []
def parse_config(self):
"""Parse SSH config file into dict"""
config = {}
try:
with open(self.config_path, 'r') as f:
for line in f:
line = line.strip()
if line and not line.startswith('#'):
parts = line.split(None, 1)
if len(parts) == 2:
config[parts[0].lower()] = parts[1]
except FileNotFoundError:
self.findings.append({
'severity': 'HIGH',
'issue': f'Config file not found: {self.config_path}'
})
return config
def check_security_settings(self):
"""Check for common SSH security issues"""
config = self.parse_config()
# Check root login
if config.get('permitrootlogin', 'yes').lower() != 'no':
self.findings.append({
'severity': 'HIGH',
'issue': 'Root login is permitted',
'recommendation': 'Set PermitRootLogin no'
})
# Check password authentication
if config.get('passwordauthentication', 'yes').lower() != 'no':
self.findings.append({
'severity': 'MEDIUM',
'issue': 'Password authentication enabled',
'recommendation': 'Use key-based auth only'
})
# Check protocol version
protocol = config.get('protocol', '2')
if '1' in protocol:
self.findings.append({
'severity': 'CRITICAL',
'issue': 'SSH Protocol 1 enabled',
'recommendation': 'Use Protocol 2 only'
})
return self.findings
Reporting and Alerting
Detection without notification is useless. Integrate your vulnerability checks with alerting systems—Slack, email, PagerDuty, or your ticketing system. Here's a simple Slack integration:
def send_slack_alert(webhook_url, findings):
"""Send vulnerability findings to Slack"""
critical = [f for f in findings if f.get('severity') == 'CRITICAL']
high = [f for f in findings if f.get('severity') == 'HIGH']
color = '#ff0000' if critical else '#ff9900' if high else '#36a64f'
message = {
'attachments': [{
'color': color,
'title': 'Vulnerability Scan Results',
'fields': [
{'title': 'Critical', 'value': str(len(critical)), 'short': True},
{'title': 'High', 'value': str(len(high)), 'short': True}
],
'text': '\n'.join([f"• {f['issue']}" for f in findings[:5]])
}]
}
requests.post(webhook_url, json=message)
Production Considerations
Moving from prototype to production requires attention to several critical areas:
Rate Limiting and Caching
External APIs have rate limits. Implement exponential backoff, respect