ffsshhttiikk What I do
- Scan for vulnerabilities in systems and applications
- Assess vulnerability severity and risk
- Prioritize remediation efforts
- Verify vulnerability fixes
- Maintain vulnerability inventory
- Provide remediation guidance
When to use me
When identifying security weaknesses in code, infrastructure, or applications.
Vulnerability Assessment Process
Discovery
from enum import Enum
class VulnerabilityType(Enum):
INJECTION = "injection"
AUTHENTICATION = "authentication"
SENSITIVE_DATA = "sensitive_data"
XXE = "xxe"
ACCESS_CONTROL = "access_control"
MISCONFIGURATION = "misconfiguration"
class VulnerabilityScanner:
"""Automated vulnerability discovery"""
def __init__(self):
self.findings = []
def scan_web_application(self, url: str) -> list:
"""Scan web application for vulnerabilities"""
vulnerabilities = []
# Check for SQL injection
if self._test_sql_injection(url):
vulnerabilities.append(self._create_finding(
"SQL Injection",
VulnerabilityType.INJECTION,
"high",
"Unsanitized input in database query"
))
# Check for XSS
if self._test_xss(url):
vulnerabilities.append(self._create_finding(
"Cross-Site Scripting",
VulnerabilityType.INJECTION,
"medium",
"Unescaped user output"
))
# Check for CSRF
if self._test_csrf(url):
vulnerabilities.append(self._create_finding(
"CSRF",
VulnerabilityType.ACCESS_CONTROL,
"medium",
"Missing anti-CSRF token"
))
return vulnerabilities
def scan_code(self, code: str, language: str) -> list:
"""Static code analysis"""
findings = []
# Check for hardcoded credentials
if "password =" in code.lower():
findings.append(self._create_finding(
"Hardcoded Credentials",
VulnerabilityType.SENSITIVE_DATA,
"critical",
"Hardcoded password in source"
))
# Check for SQL injection risk
if "execute(" in code and "f'" in code:
findings.append(self._create_finding(
"Potential SQL Injection",
VulnerabilityType.INJECTION,
"high",
"String formatting in SQL query"
))
return findings
def _create_finding(self, name: str, vuln_type: VulnerabilityType,
severity: str, description: str) -> dict:
return {
"name": name,
"type": vuln_type.value,
"severity": severity,
"description": description,
"cwe": self._lookup_cwe(name),
"cvss": self._calculate_cvss(severity)
}
Assessment
class VulnerabilityAssessment:
"""Assess and prioritize vulnerabilities"""
CVSS_SEVERITY = {
9.0: "critical",
7.0: "high",
4.0: "medium",
0.1: "low",
0.0: "info"
}
def __init__(self):
self.vulnerabilities = []
def add_vulnerability(self, vuln: dict):
"""Add vulnerability to assessment"""
vuln["cvss_score"] = self._calculate_cvss_score(vuln)
vuln["risk_level"] = self._determine_risk_level(vuln["cvss_score"])
vuln["affected_assets"] = []
vuln["remediation_status"] = "open"
self.vulnerabilities.append(vuln)
def _calculate_cvss_score(self, vuln: dict) -> float:
"""Calculate CVSS 3.1 score"""
# Simplified CVSS calculation
base_score = 0
if vuln.get("severity") == "critical":
base_score = 9.5
elif vuln.get("severity") == "high":
base_score = 7.5
elif vuln.get("severity") == "medium":
base_score = 5.0
else:
base_score = 2.5
# Adjust for exploitability
if vuln.get("exploitable"):
base_score = min(base_score + 0.5, 10.0)
return base_score
def prioritize(self) -> list:
"""Sort vulnerabilities by priority"""
return sorted(
self.vulnerabilities,
key=lambda v: (v["risk_level"], v["cvss_score"]),
reverse=True
)
def get_remediation_sla(self, risk_level: str) -> int:
"""Get remediation SLA in days"""
return {
"critical": 7,
"high": 30,
"medium": 90,
"low": 180
}.get(risk_level, 180)
Remediation Tracking
class RemediationTracker:
"""Track vulnerability remediation"""
def __init__(self):
self.remediation_queue = []
def create_remediation_task(self, vulnerability: dict,
assignee: str) -> dict:
"""Create remediation task"""
task = {
"id": f"REM-{len(self.remediation_queue) + 1}",
"vulnerability": vulnerability,
"assignee": assignee,
"status": "in_progress",
"created": datetime.now(),
"due_date": datetime.now() + timedelta(
days=vulnerability.get("sla_days", 30)
),
"steps": self._generate_remediation_steps(vulnerability)
}
self.remediation_queue.append(task)
return task
def _generate_remediation_steps(self, vuln: dict) -> list:
"""Generate remediation steps based on vulnerability type"""
templates = {
"sql_injection": [
"Use parameterized queries",
"Implement input validation",
"Apply least privilege to database user"
],
"xss": [
"Escape output based on context",
"Use Content Security Policy",
"Implement output encoding"
],
"hardcoded_credentials": [
"Remove credentials from code",
"Implement secrets management",
"Rotate exposed credentials"
]
}
return templates.get(vuln.get("type"), ["Investigate and fix"])
def verify_remediation(self, task_id: str) -> bool:
"""Verify vulnerability is fixed"""
task = next((t for t in self.remediation_queue if t["id"] == task_id), None)
if not task:
return False
# Re-scan to verify fix
# In practice, this would trigger a rescan
task["status"] = "verified"
task["verified_date"] = datetime.now()
return True
OWASP Top 10
class OWASPAssessment:
"""OWASP Top 10 vulnerability checks"""
TOP_10 = [
"A01:2021-Broken Access Control",
"A02:2021-Cryptographic Failures",
"A03:2021-Injection",
"A04:2021-Insecure Design",
"A05:2021-Security Misconfiguration",
"A06:2021-Vulnerable Components",
"A07:2021-Authentication Failures",
"A08:2021-Software Data Integrity Failures",
"A09:2021-Security Logging Failures",
"A10:2021-Server-Side Request Forgery"
]
def check_access_control(self, app) -> dict:
"""Check for broken access control"""
checks = [
{
"check": "Directory traversal",
"vulnerable": self._test_directory_traversal(app),
"remediation": "Validate and sanitize file paths"
},
{
"check": "IDOR",
"vulnerable": self._test_idor(app),
"remediation": "Implement object-level authorization"
},
{
"check": "Privilege escalation",
"vulnerable": self._test_privilege_escalation(app),
"remediation": "Verify user permissions on each request"
}
]
return {
"category": self.TOP_10[0],
"findings": [c for c in checks if c["vulnerable"]],
"overall_status": "pass" if not any(c["vulnerable"] for c in checks) else "fail"
}