Security Automation

1import subprocess
2import json
3from typing import Dict, List, Optional
4from dataclasses import dataclass
5from enum import Enum
6import logging
7import requests
8
9logger = logging.getLogger(__name__)
10
11class VulnerabilitySeverity(Enum):
12    """Vulnerability severity levels"""
13    CRITICAL = "critical"
14    HIGH = "high"
15    MEDIUM = "medium"
16    LOW = "low"
17    INFO = "info"
18
19@dataclass
20class Vulnerability:
21    """Vulnerability details"""
22    package: str
23    version: str
24    vulnerability_id: str  # CVE-2024-1234
25    severity: VulnerabilitySeverity
26    description: str
27    fixed_version: Optional[str] = None
28    cvss_score: Optional[float] = None
29    published_date: Optional[str] = None
30
31class DependencyScanner:
32    """
33    Scan Python dependencies for known vulnerabilities
34    """
35
36    def __init__(self):
37        """Initialize dependency scanner"""
38        logger.info("🔍 Initializing dependency scanner")
39
40    def scan_requirements(self, requirements_file: str = 'requirements.txt') -> List[Vulnerability]:
41        """
42        Scan requirements.txt for vulnerabilities using Safety
43
44        Args:
45            requirements_file: Path to requirements.txt
46
47        Returns:
48            List of vulnerabilities found
49        """
50        try:
51            logger.info(f"📋 Scanning {requirements_file} for vulnerabilities...")
52
53            # Run safety check (requires: pip install safety)
54            result = subprocess.run(
55                ['safety', 'check', '--file', requirements_file, '--json'],
56                capture_output=True,
57                text=True,
58                timeout=60
59            )
60
61            vulnerabilities = []
62
63            if result.returncode != 0:
64                # Parse JSON output
65                try:
66                    safety_output = json.loads(result.stdout)
67
68                    for vuln_data in safety_output:
69                        vulnerability = Vulnerability(
70                            package=vuln_data[0],
71                            version=vuln_data[2],
72                            vulnerability_id=vuln_data[3],
73                            severity=self._map_severity(vuln_data[4]),
74                            description=vuln_data[1],
75                            fixed_version=vuln_data[5] if len(vuln_data) > 5 else None
76                        )
77                        vulnerabilities.append(vulnerability)
78
79                except json.JSONDecodeError:
80                    logger.warning("Failed to parse safety output")
81
82            logger.info(f"✅ Found {len(vulnerabilities)} vulnerabilities")
83
84            return vulnerabilities
85
86        except subprocess.TimeoutExpired:
87            logger.error("❌ Safety check timed out")
88            return []
89        except FileNotFoundError:
90            logger.error("❌ Safety tool not found. Install: pip install safety")
91            return []
92        except Exception as e:
93            logger.error(f"❌ Scan failed: {e}")
94            return []
95
96    def scan_installed_packages(self) -> List[Vulnerability]:
97        """
98        Scan currently installed packages
99
100        Returns:
101            List of vulnerabilities in installed packages
102        """
103        try:
104            logger.info("📦 Scanning installed packages...")
105
106            # Run safety check on installed packages
107            result = subprocess.run(
108                ['safety', 'check', '--json'],
109                capture_output=True,
110                text=True,
111                timeout=60
112            )
113
114            vulnerabilities = []
115
116            if result.returncode != 0:
117                try:
118                    safety_output = json.loads(result.stdout)
119
120                    for vuln_data in safety_output:
121                        vulnerability = Vulnerability(
122                            package=vuln_data[0],
123                            version=vuln_data[2],
124                            vulnerability_id=vuln_data[3],
125                            severity=self._map_severity(vuln_data[4]),
126                            description=vuln_data[1]
127                        )
128                        vulnerabilities.append(vulnerability)
129
130                except json.JSONDecodeError:
131                    pass
132
133            logger.info(f"✅ Found {len(vulnerabilities)} vulnerabilities in installed packages")
134
135            return vulnerabilities
136
137        except Exception as e:
138            logger.error(f"❌ Scan failed: {e}")
139            return []
140
141    def check_cve_database(self, package: str, version: str) -> List[Dict]:
142        """
143        Check CVE database for package vulnerabilities
144
145        Args:
146            package: Package name
147            version: Package version
148
149        Returns:
150            List of CVEs found
151        """
152        try:
153            # Query OSV (Open Source Vulnerabilities) database
154            url = "https://api.osv.dev/v1/query"
155
156            payload = {
157                "package": {
158                    "name": package,
159                    "ecosystem": "PyPI"
160                },
161                "version": version
162            }
163
164            response = requests.post(url, json=payload, timeout=10)
165
166            if response.status_code == 200:
167                data = response.json()
168                vulns = data.get('vulns', [])
169
170                logger.info(f"✅ Found {len(vulns)} CVEs for {package}=={version}")
171
172                return vulns
173            else:
174                logger.warning(f"⚠️  CVE check failed: {response.status_code}")
175                return []
176
177        except requests.RequestException as e:
178            logger.error(f"❌ CVE database query failed: {e}")
179            return []
180
181    def _map_severity(self, severity_str: str) -> VulnerabilitySeverity:
182        """Map severity string to enum"""
183        severity_map = {
184            'critical': VulnerabilitySeverity.CRITICAL,
185            'high': VulnerabilitySeverity.HIGH,
186            'medium': VulnerabilitySeverity.MEDIUM,
187            'low': VulnerabilitySeverity.LOW
188        }
189
190        return severity_map.get(severity_str.lower(), VulnerabilitySeverity.INFO)
191
192    def generate_report(self, vulnerabilities: List[Vulnerability]) -> Dict:
193        """
194        Generate vulnerability report
195
196        Args:
197            vulnerabilities: List of vulnerabilities
198
199        Returns:
200            Report summary
201        """
202        report = {
203            'total_vulnerabilities': len(vulnerabilities),
204            'by_severity': {
205                'critical': 0,
206                'high': 0,
207                'medium': 0,
208                'low': 0,
209                'info': 0
210            },
211            'vulnerable_packages': set(),
212            'critical_vulnerabilities': []
213        }
214
215        for vuln in vulnerabilities:
216            # Count by severity
217            report['by_severity'][vuln.severity.value] += 1
218
219            # Track vulnerable packages
220            report['vulnerable_packages'].add(f"{vuln.package}=={vuln.version}")
221
222            # Track critical vulnerabilities
223            if vuln.severity == VulnerabilitySeverity.CRITICAL:
224                report['critical_vulnerabilities'].append({
225                    'package': vuln.package,
226                    'version': vuln.version,
227                    'cve': vuln.vulnerability_id,
228                    'description': vuln.description,
229                    'fixed_version': vuln.fixed_version
230                })
231
232        report['vulnerable_packages'] = list(report['vulnerable_packages'])
233
234        return report

1import subprocess
2from pathlib import Path
3from typing import Dict, List
4import json
5
6class CodeSecurityScanner:
7    """
8    Static Application Security Testing (SAST)
9    """
10
11    def __init__(self, project_path: str = '.'):
12        """Initialize code scanner"""
13        self.project_path = Path(project_path)
14        logger.info(f"🔒 Initializing code security scanner for {project_path}")
15
16    def scan_with_bandit(self) -> Dict:
17        """
18        Scan Python code for security issues using Bandit
19
20        Returns:
21            Security issues found
22        """
23        try:
24            logger.info("🔍 Running Bandit security scan...")
25
26            # Run bandit (requires: pip install bandit)
27            result = subprocess.run(
28                ['bandit', '-r', str(self.project_path), '-f', 'json'],
29                capture_output=True,
30                text=True,
31                timeout=300
32            )
33
34            issues = []
35
36            if result.stdout:
37                try:
38                    bandit_output = json.loads(result.stdout)
39
40                    for issue in bandit_output.get('results', []):
41                        issues.append({
42                            'file': issue['filename'],
43                            'line': issue['line_number'],
44                            'severity': issue['issue_severity'],
45                            'confidence': issue['issue_confidence'],
46                            'issue': issue['issue_text'],
47                            'code': issue['code']
48                        })
49
50                except json.JSONDecodeError:
51                    logger.warning("Failed to parse Bandit output")
52
53            logger.info(f"✅ Bandit found {len(issues)} security issues")
54
55            return {
56                'tool': 'bandit',
57                'issues_found': len(issues),
58                'issues': issues
59            }
60
61        except subprocess.TimeoutExpired:
62            logger.error("❌ Bandit scan timed out")
63            return {'tool': 'bandit', 'issues_found': 0, 'error': 'timeout'}
64        except FileNotFoundError:
65            logger.error("❌ Bandit not found. Install: pip install bandit")
66            return {'tool': 'bandit', 'issues_found': 0, 'error': 'not_installed'}
67        except Exception as e:
68            logger.error(f"❌ Bandit scan failed: {e}")
69            return {'tool': 'bandit', 'issues_found': 0, 'error': str(e)}
70
71    def scan_secrets(self) -> Dict:
72        """
73        Scan for exposed secrets and credentials
74
75        Returns:
76            Exposed secrets found
77        """
78        try:
79            logger.info("🔑 Scanning for exposed secrets...")
80
81            # Patterns for common secrets
82            secret_patterns = {
83                'aws_access_key': r'AKIA[0-9A-Z]{16}',
84                'aws_secret_key': r'aws(.{0,20})?['\"\']?[0-9a-zA-Z\/+]{40}['\"\']?',
85                'github_token': r'ghp_[0-9a-zA-Z]{36}',
86                'slack_token': r'xox[baprs]-[0-9]{10,12}-[0-9]{10,12}-[0-9a-zA-Z]{24,32}',
87                'private_key': r'-----BEGIN (RSA|DSA|EC|OPENSSH) PRIVATE KEY-----',
88                'generic_api_key': r'api[_-]?key[_-]?([\s:]?=?[\s:]?['\"\']?)[0-9a-zA-Z]{32,45}['\"\']?',
89                'password': r'password[_-]?([\s:]?=?[\s:]?['\"\']?)[0-9a-zA-Z]{8,}['\"\']?'
90            }
91
92            secrets_found = []
93
94            # Scan all Python files
95            for py_file in self.project_path.rglob('*.py'):
96                try:
97                    content = py_file.read_text(encoding='utf-8')
98
99                    for secret_type, pattern in secret_patterns.items():
100                        import re
101                        matches = re.finditer(pattern, content, re.IGNORECASE)
102
103                        for match in matches:
104                            # Get line number
105                            line_num = content[:match.start()].count('\n') + 1
106
107                            secrets_found.append({
108                                'file': str(py_file.relative_to(self.project_path)),
109                                'line': line_num,
110                                'type': secret_type,
111                                'match': match.group()[:50] + '...'  # Truncate for safety
112                            })
113
114                except Exception as e:
115                    logger.debug(f"Could not scan {py_file}: {e}")
116
117            logger.info(f"✅ Found {len(secrets_found)} potential secrets")
118
119            return {
120                'tool': 'secrets_scanner',
121                'secrets_found': len(secrets_found),
122                'secrets': secrets_found
123            }
124
125        except Exception as e:
126            logger.error(f"❌ Secrets scan failed: {e}")
127            return {'tool': 'secrets_scanner', 'secrets_found': 0, 'error': str(e)}

1import subprocess
2import json
3from typing import Dict, List
4from enum import Enum
5import logging
6
7logger = logging.getLogger(__name__)
8
9class ComplianceFramework(Enum):
10    """Security compliance frameworks"""
11    CIS_BENCHMARK = "cis"
12    PCI_DSS = "pci-dss"
13    HIPAA = "hipaa"
14    SOC2 = "soc2"
15    GDPR = "gdpr"
16    NIST = "nist"
17
18@dataclass
19class ComplianceCheck:
20    """Compliance check result"""
21    check_id: str
22    title: str
23    status: str  # pass, fail, warning
24    severity: str
25    description: str
26    remediation: str
27    framework: ComplianceFramework
28
29class CloudSecurityScanner:
30    """
31    Scan cloud infrastructure for security compliance
32    """
33
34    def __init__(self, cloud_provider: str = 'aws'):
35        """Initialize cloud security scanner"""
36        self.cloud_provider = cloud_provider
37        logger.info(f"☁️  Initializing {cloud_provider.upper()} security scanner")
38
39    def scan_aws_security(self) -> Dict:
40        """
41        Scan AWS infrastructure using Prowler
42
43        Returns:
44            Security findings
45        """
46        try:
47            logger.info("🔍 Running AWS security assessment with Prowler...")
48
49            # Run prowler (requires: pip install prowler)
50            # Note: In production, specify profile and regions
51            result = subprocess.run(
52                ['prowler', '--output-modes', 'json'],
53                capture_output=True,
54                text=True,
55                timeout=600
56            )
57
58            findings = []
59
60            if result.stdout:
61                try:
62                    # Parse prowler JSON output
63                    prowler_output = json.loads(result.stdout)
64
65                    for finding in prowler_output.get('findings', []):
66                        findings.append({
67                            'check_id': finding.get('CheckID'),
68                            'title': finding.get('CheckTitle'),
69                            'severity': finding.get('Severity'),
70                            'status': finding.get('Status'),
71                            'resource': finding.get('ResourceId'),
72                            'region': finding.get('Region'),
73                            'service': finding.get('Service'),
74                            'description': finding.get('Description'),
75                            'remediation': finding.get('Remediation')
76                        })
77
78                except json.JSONDecodeError:
79                    logger.warning("Failed to parse Prowler output")
80
81            logger.info(f"✅ AWS scan found {len(findings)} security findings")
82
83            return {
84                'provider': 'aws',
85                'tool': 'prowler',
86                'findings_count': len(findings),
87                'findings': findings
88            }
89
90        except subprocess.TimeoutExpired:
91            logger.error("❌ AWS scan timed out")
92            return {'provider': 'aws', 'findings_count': 0, 'error': 'timeout'}
93        except FileNotFoundError:
94            logger.error("❌ Prowler not found. Install: pip install prowler")
95            return {'provider': 'aws', 'findings_count': 0, 'error': 'not_installed'}
96        except Exception as e:
97            logger.error(f"❌ AWS scan failed: {e}")
98            return {'provider': 'aws', 'findings_count': 0, 'error': str(e)}
99
100    def check_s3_bucket_security(self, bucket_name: str) -> List[ComplianceCheck]:
101        """
102        Check S3 bucket security configuration
103
104        Args:
105            bucket_name: S3 bucket name
106
107        Returns:
108            List of compliance checks
109        """
110        import boto3
111        from botocore.exceptions import ClientError
112
113        checks = []
114
115        try:
116            s3_client = boto3.client('s3')
117
118            # Check 1: Bucket encryption
119            try:
120                encryption = s3_client.get_bucket_encryption(Bucket=bucket_name)
121                checks.append(ComplianceCheck(
122                    check_id='S3.1',
123                    title='S3 bucket encryption enabled',
124                    status='pass',
125                    severity='high',
126                    description=f'Bucket {bucket_name} has encryption enabled',
127                    remediation='N/A - Already compliant',
128                    framework=ComplianceFramework.CIS_BENCHMARK
129                ))
130            except ClientError:
131                checks.append(ComplianceCheck(
132                    check_id='S3.1',
133                    title='S3 bucket encryption enabled',
134                    status='fail',
135                    severity='high',
136                    description=f'Bucket {bucket_name} does not have encryption enabled',
137                    remediation='Enable default encryption on S3 bucket',
138                    framework=ComplianceFramework.CIS_BENCHMARK
139                ))
140
141            # Check 2: Public access block
142            try:
143                public_access = s3_client.get_public_access_block(Bucket=bucket_name)
144                config = public_access['PublicAccessBlockConfiguration']
145
146                if all([
147                    config.get('BlockPublicAcls'),
148                    config.get('IgnorePublicAcls'),
149                    config.get('BlockPublicPolicy'),
150                    config.get('RestrictPublicBuckets')
151                ]):
152                    checks.append(ComplianceCheck(
153                        check_id='S3.2',
154                        title='S3 bucket public access blocked',
155                        status='pass',
156                        severity='critical',
157                        description=f'Bucket {bucket_name} blocks all public access',
158                        remediation='N/A - Already compliant',
159                        framework=ComplianceFramework.CIS_BENCHMARK
160                    ))
161                else:
162                    checks.append(ComplianceCheck(
163                        check_id='S3.2',
164                        title='S3 bucket public access blocked',
165                        status='fail',
166                        severity='critical',
167                        description=f'Bucket {bucket_name} allows some public access',
168                        remediation='Enable all public access block settings',
169                        framework=ComplianceFramework.CIS_BENCHMARK
170                    ))
171
172            except ClientError:
173                checks.append(ComplianceCheck(
174                    check_id='S3.2',
175                    title='S3 bucket public access blocked',
176                    status='fail',
177                    severity='critical',
178                    description=f'Bucket {bucket_name} has no public access block',
179                    remediation='Configure public access block for bucket',
180                    framework=ComplianceFramework.CIS_BENCHMARK
181                ))
182
183            # Check 3: Versioning
184            try:
185                versioning = s3_client.get_bucket_versioning(Bucket=bucket_name)
186                if versioning.get('Status') == 'Enabled':
187                    checks.append(ComplianceCheck(
188                        check_id='S3.3',
189                        title='S3 bucket versioning enabled',
190                        status='pass',
191                        severity='medium',
192                        description=f'Bucket {bucket_name} has versioning enabled',
193                        remediation='N/A - Already compliant',
194                        framework=ComplianceFramework.CIS_BENCHMARK
195                    ))
196                else:
197                    checks.append(ComplianceCheck(
198                        check_id='S3.3',
199                        title='S3 bucket versioning enabled',
200                        status='fail',
201                        severity='medium',
202                        description=f'Bucket {bucket_name} does not have versioning',
203                        remediation='Enable versioning on S3 bucket',
204                        framework=ComplianceFramework.CIS_BENCHMARK
205                    ))
206            except ClientError:
207                pass
208
209            # Check 4: Access logging
210            try:
211                logging_config = s3_client.get_bucket_logging(Bucket=bucket_name)
212                if 'LoggingEnabled' in logging_config:
213                    checks.append(ComplianceCheck(
214                        check_id='S3.4',
215                        title='S3 bucket access logging enabled',
216                        status='pass',
217                        severity='medium',
218                        description=f'Bucket {bucket_name} has access logging',
219                        remediation='N/A - Already compliant',
220                        framework=ComplianceFramework.CIS_BENCHMARK
221                    ))
222                else:
223                    checks.append(ComplianceCheck(
224                        check_id='S3.4',
225                        title='S3 bucket access logging enabled',
226                        status='fail',
227                        severity='medium',
228                        description=f'Bucket {bucket_name} has no access logging',
229                        remediation='Enable server access logging for bucket',
230                        framework=ComplianceFramework.CIS_BENCHMARK
231                    ))
232            except ClientError:
233                pass
234
235        except Exception as e:
236            logger.error(f"❌ S3 security check failed: {e}")
237
238        return checks
239
240    def check_ec2_security(self) -> List[ComplianceCheck]:
241        """
242        Check EC2 instance security
243
244        Returns:
245            List of compliance checks
246        """
247        import boto3
248
249        checks = []
250
251        try:
252            ec2_client = boto3.client('ec2')
253
254            # Get all security groups
255            security_groups = ec2_client.describe_security_groups()
256
257            for sg in security_groups['SecurityGroups']:
258                # Check for overly permissive rules
259                for rule in sg.get('IpPermissions', []):
260                    for ip_range in rule.get('IpRanges', []):
261                        if ip_range.get('CidrIp') == '0.0.0.0/0':
262                            # Check if allows unrestricted access
263                            from_port = rule.get('FromPort', 0)
264                            to_port = rule.get('ToPort', 65535)
265
266                            # Critical: SSH (22) or RDP (3389) open to world
267                            if from_port <= 22 <= to_port or from_port <= 3389 <= to_port:
268                                checks.append(ComplianceCheck(
269                                    check_id='EC2.1',
270                                    title='Security group restricts public access',
271                                    status='fail',
272                                    severity='critical',
273                                    description=f'Security group {sg["GroupName"]} allows unrestricted SSH/RDP',
274                                    remediation='Restrict SSH/RDP access to specific IP ranges',
275                                    framework=ComplianceFramework.CIS_BENCHMARK
276                                ))
277
278            # Get all EC2 instances
279            instances = ec2_client.describe_instances()
280
281            for reservation in instances['Reservations']:
282                for instance in reservation['Instances']:
283                    instance_id = instance['InstanceId']
284
285                    # Check if instance has public IP
286                    if instance.get('PublicIpAddress'):
287                        checks.append(ComplianceCheck(
288                            check_id='EC2.2',
289                            title='EC2 instance uses private IP',
290                            status='warning',
291                            severity='medium',
292                            description=f'Instance {instance_id} has public IP',
293                            remediation='Consider using private subnets with NAT gateway',
294                            framework=ComplianceFramework.CIS_BENCHMARK
295                        ))
296
297                    # Check if IMDSv2 is enforced
298                    metadata_options = instance.get('MetadataOptions', {})
299                    if metadata_options.get('HttpTokens') != 'required':
300                        checks.append(ComplianceCheck(
301                            check_id='EC2.3',
302                            title='EC2 instance enforces IMDSv2',
303                            status='fail',
304                            severity='high',
305                            description=f'Instance {instance_id} does not enforce IMDSv2',
306                            remediation='Require IMDSv2 for instance metadata access',
307                            framework=ComplianceFramework.CIS_BENCHMARK
308                        ))
309
310        except Exception as e:
311            logger.error(f"❌ EC2 security check failed: {e}")
312
313        return checks
314
315    def generate_compliance_report(self, checks: List[ComplianceCheck]) -> Dict:
316        """
317        Generate compliance report
318
319        Args:
320            checks: List of compliance checks
321
322        Returns:
323            Compliance report
324        """
325        report = {
326            'total_checks': len(checks),
327            'passed': 0,
328            'failed': 0,
329            'warnings': 0,
330            'by_severity': {
331                'critical': 0,
332                'high': 0,
333                'medium': 0,
334                'low': 0
335            },
336            'by_framework': {},
337            'failed_checks': []
338        }
339
340        for check in checks:
341            # Count by status
342            if check.status == 'pass':
343                report['passed'] += 1
344            elif check.status == 'fail':
345                report['failed'] += 1
346                report['failed_checks'].append({
347                    'id': check.check_id,
348                    'title': check.title,
349                    'severity': check.severity,
350                    'remediation': check.remediation
351                })
352            elif check.status == 'warning':
353                report['warnings'] += 1
354
355            # Count by severity
356            report['by_severity'][check.severity] += 1
357
358            # Count by framework
359            framework = check.framework.value
360            report['by_framework'][framework] = report['by_framework'].get(framework, 0) + 1
361
362        # Calculate compliance score
363        if report['total_checks'] > 0:
364            report['compliance_score'] = (report['passed'] / report['total_checks']) * 100
365        else:
366            report['compliance_score'] = 0
367
368        return report

1class DockerSecurityScanner:
2    """Scan Docker images for vulnerabilities"""
3
4    def scan_image(self, image_name: str) -> Dict:
5        """
6        Scan Docker image with Trivy
7
8        Args:
9            image_name: Docker image name
10
11        Returns:
12            Scan results
13        """
14        try:
15            logger.info(f"🐳 Scanning Docker image: {image_name}")
16
17            # Run trivy (requires: trivy installation)
18            result = subprocess.run(
19                ['trivy', 'image', '--format', 'json', image_name],
20                capture_output=True,
21                text=True,
22                timeout=300
23            )
24
25            vulnerabilities = []
26
27            if result.stdout:
28                try:
29                    trivy_output = json.loads(result.stdout)
30
31                    for result_item in trivy_output.get('Results', []):
32                        for vuln in result_item.get('Vulnerabilities', []):
33                            vulnerabilities.append({
34                                'cve': vuln.get('VulnerabilityID'),
35                                'package': vuln.get('PkgName'),
36                                'version': vuln.get('InstalledVersion'),
37                                'severity': vuln.get('Severity'),
38                                'fixed_version': vuln.get('FixedVersion'),
39                                'title': vuln.get('Title')
40                            })
41
42                except json.JSONDecodeError:
43                    logger.warning("Failed to parse Trivy output")
44
45            logger.info(f"✅ Found {len(vulnerabilities)} vulnerabilities in image")
46
47            return {
48                'image': image_name,
49                'vulnerabilities_count': len(vulnerabilities),
50                'vulnerabilities': vulnerabilities
51            }
52
53        except subprocess.TimeoutExpired:
54            logger.error("❌ Image scan timed out")
55            return {'image': image_name, 'vulnerabilities_count': 0, 'error': 'timeout'}
56        except FileNotFoundError:
57            logger.error("❌ Trivy not found")
58            return {'image': image_name, 'vulnerabilities_count': 0, 'error': 'not_installed'}
59        except Exception as e:
60            logger.error(f"❌ Image scan failed: {e}")
61            return {'image': image_name, 'vulnerabilities_count': 0, 'error': str(e)}

1from typing import Dict, List, Callable
2from datetime import datetime
3from enum import Enum
4from dataclasses import dataclass
5import logging
6
7logger = logging.getLogger(__name__)
8
9class ThreatLevel(Enum):
10    """Threat severity levels"""
11    CRITICAL = "critical"
12    HIGH = "high"
13    MEDIUM = "medium"
14    LOW = "low"
15    INFO = "info"
16
17@dataclass
18class SecurityEvent:
19    """Security event details"""
20    event_id: str
21    timestamp: datetime
22    event_type: str
23    threat_level: ThreatLevel
24    source_ip: str
25    target: str
26    description: str
27    indicators: Dict
28    recommended_action: str
29
30class SecurityMonitor:
31    """
32    Monitor security events and detect threats
33    """
34
35    def __init__(self):
36        """Initialize security monitor"""
37        self.event_handlers: List[Callable] = []
38        self.threat_rules = self._initialize_threat_rules()
39
40        logger.info("🛡️  Security monitor initialized")
41
42    def _initialize_threat_rules(self) -> Dict:
43        """Initialize threat detection rules"""
44        return {
45            'brute_force': {
46                'threshold': 5,
47                'timewindow': 60,  # seconds
48                'threat_level': ThreatLevel.HIGH
49            },
50            'sql_injection': {
51                'patterns': ['UNION SELECT', 'DROP TABLE', '--', 'OR 1=1'],
52                'threat_level': ThreatLevel.CRITICAL
53            },
54            'xss_attempt': {
55                'patterns': ['<script>', 'javascript:', 'onerror='],
56                'threat_level': ThreatLevel.HIGH
57            },
58            'suspicious_file_access': {
59                'files': ['/etc/passwd', '/etc/shadow', '/.ssh/'],
60                'threat_level': ThreatLevel.CRITICAL
61            }
62        }
63
64    def analyze_failed_logins(self, log_entries: List[Dict]) -> List[SecurityEvent]:
65        """
66        Analyze failed login attempts for brute force attacks
67
68        Args:
69            log_entries: List of authentication log entries
70
71        Returns:
72            List of security events
73        """
74        events = []
75        failed_attempts = {}
76
77        for entry in log_entries:
78            if entry.get('event_type') == 'failed_login':
79                source_ip = entry.get('source_ip')
80                timestamp = entry.get('timestamp')
81
82                if source_ip not in failed_attempts:
83                    failed_attempts[source_ip] = []
84
85                failed_attempts[source_ip].append(timestamp)
86
87        # Check for brute force patterns
88        rule = self.threat_rules['brute_force']
89
90        for ip, attempts in failed_attempts.items():
91            if len(attempts) >= rule['threshold']:
92                event = SecurityEvent(
93                    event_id=f"BF-{ip}-{datetime.now().timestamp()}",
94                    timestamp=datetime.now(),
95                    event_type='brute_force_attempt',
96                    threat_level=rule['threat_level'],
97                    source_ip=ip,
98                    target='authentication_system',
99                    description=f'Detected {len(attempts)} failed login attempts from {ip}',
100                    indicators={'failed_attempts': len(attempts), 'timespan': rule['timewindow']},
101                    recommended_action=f'Block IP {ip} and investigate user accounts targeted'
102                )
103
104                events.append(event)
105                logger.warning(f"🚨 Brute force detected from {ip}: {len(attempts)} attempts")
106
107        return events
108
109    def detect_sql_injection(self, web_requests: List[Dict]) -> List[SecurityEvent]:
110        """
111        Detect SQL injection attempts in web requests
112
113        Args:
114            web_requests: List of HTTP request logs
115
116        Returns:
117            List of security events
118        """
119        events = []
120        rule = self.threat_rules['sql_injection']
121
122        for request in web_requests:
123            query_string = request.get('query_string', '')
124            body = request.get('body', '')
125            combined = f"{query_string} {body}".upper()
126
127            for pattern in rule['patterns']:
128                if pattern in combined:
129                    event = SecurityEvent(
130                        event_id=f"SQLI-{request.get('request_id', 'unknown')}",
131                        timestamp=datetime.now(),
132                        event_type='sql_injection_attempt',
133                        threat_level=rule['threat_level'],
134                        source_ip=request.get('source_ip', 'unknown'),
135                        target=request.get('endpoint', 'unknown'),
136                        description=f'SQL injection pattern detected: {pattern}',
137                        indicators={'pattern': pattern, 'endpoint': request.get('endpoint')},
138                        recommended_action='Block request, review application input validation'
139                    )
140
141                    events.append(event)
142                    logger.critical(f"🚨 SQL injection attempt detected: {pattern}")
143                    break
144
145        return events
146
147    def monitor_file_access(self, file_access_logs: List[Dict]) -> List[SecurityEvent]:
148        """
149        Monitor for suspicious file access
150
151        Args:
152            file_access_logs: List of file access events
153
154        Returns:
155            List of security events
156        """
157        events = []
158        rule = self.threat_rules['suspicious_file_access']
159
160        for log in file_access_logs:
161            file_path = log.get('file_path', '')
162
163            for sensitive_file in rule['files']:
164                if sensitive_file in file_path:
165                    event = SecurityEvent(
166                        event_id=f"FILE-{log.get('user', 'unknown')}-{datetime.now().timestamp()}",
167                        timestamp=datetime.now(),
168                        event_type='suspicious_file_access',
169                        threat_level=rule['threat_level'],
170                        source_ip=log.get('source_ip', 'localhost'),
171                        target=file_path,
172                        description=f'Suspicious access to {file_path} by {log.get("user")}',
173                        indicators={'file': file_path, 'user': log.get('user'), 'action': log.get('action')},
174                        recommended_action='Investigate user activity and review file permissions'
175                    )
176
177                    events.append(event)
178                    logger.critical(f"🚨 Suspicious file access: {file_path}")
179
180        return events
181
182    def register_event_handler(self, handler: Callable):
183        """Register security event handler"""
184        self.event_handlers.append(handler)
185
186    def trigger_event(self, event: SecurityEvent):
187        """Trigger security event and call handlers"""
188        logger.warning(f"🚨 Security event: {event.event_type} - {event.description}")
189
190        for handler in self.event_handlers:
191            try:
192                handler(event)
193            except Exception as e:
194                logger.error(f"❌ Event handler failed: {e}")
195
196
197class IncidentResponseAutomation:
198    """
199    Automate incident response workflows
200    """
201
202    def __init__(self):
203        """Initialize incident response automation"""
204        logger.info("🚨 Incident response automation initialized")
205
206    def respond_to_brute_force(self, event: SecurityEvent):
207        """
208        Automated response to brute force attack
209
210        Args:
211            event: Security event
212        """
213        logger.info(f"🛡️  Responding to brute force from {event.source_ip}")
214
215        # 1. Block IP at firewall
216        self._block_ip(event.source_ip)
217
218        # 2. Alert security team
219        self._send_alert(event)
220
221        # 3. Create incident ticket
222        self._create_incident_ticket(event)
223
224        # 4. Log incident
225        self._log_incident(event)
226
227    def respond_to_sql_injection(self, event: SecurityEvent):
228        """
229        Automated response to SQL injection attempt
230
231        Args:
232            event: Security event
233        """
234        logger.critical(f"🛡️  Responding to SQL injection attempt")
235
236        # 1. Block source IP immediately
237        self._block_ip(event.source_ip)
238
239        # 2. Alert security and development teams
240        self._send_critical_alert(event)
241
242        # 3. Trigger WAF rule update
243        self._update_waf_rules(event)
244
245        # 4. Create high-priority incident
246        self._create_incident_ticket(event, priority='critical')
247
248        # 5. Capture forensics data
249        self._capture_forensics(event)
250
251    def _block_ip(self, ip: str):
252        """Block IP address"""
253        logger.info(f"🚫 Blocking IP: {ip}")
254
255        # In production, integrate with firewall/WAF
256        # import boto3
257        # waf = boto3.client('wafv2')
258        # waf.update_ip_set(...)
259
260    def _send_alert(self, event: SecurityEvent):
261        """Send security alert"""
262        logger.info(f"📧 Sending alert for event: {event.event_id}")
263
264        # In production, integrate with PagerDuty, Slack, email
265        alert_message = f"""
266        Security Alert: {event.event_type}
267        Severity: {event.threat_level.value}
268        Source: {event.source_ip}
269        Description: {event.description}
270        Action: {event.recommended_action}
271        """
272
273        # Send to alerting system
274        logger.info(alert_message)
275
276    def _send_critical_alert(self, event: SecurityEvent):
277        """Send critical security alert"""
278        logger.critical(f"🚨 CRITICAL ALERT: {event.event_type}")
279
280        # Escalate to on-call team immediately
281        # trigger_pagerduty_incident(event)
282        # post_to_security_slack_channel(event)
283
284    def _create_incident_ticket(self, event: SecurityEvent, priority: str = 'high'):
285        """Create incident ticket"""
286        logger.info(f"🎫 Creating incident ticket for: {event.event_id}")
287
288        # In production, integrate with Jira, ServiceNow
289        ticket_data = {
290            'title': f"Security Incident: {event.event_type}",
291            'description': event.description,
292            'severity': event.threat_level.value,
293            'priority': priority,
294            'assigned_to': 'security-team'
295        }
296
297        logger.info(f"Ticket created: {ticket_data}")
298
299    def _update_waf_rules(self, event: SecurityEvent):
300        """Update WAF rules based on attack pattern"""
301        logger.info(f"🔧 Updating WAF rules for: {event.event_type}")
302
303        # Add blocking rule for attack pattern
304        # waf.create_rule_group(...)
305
306    def _capture_forensics(self, event: SecurityEvent):
307        """Capture forensics data for investigation"""
308        logger.info(f"🔬 Capturing forensics data for: {event.event_id}")
309
310        # Capture network traffic, logs, system state
311        # tcpdump, memory dump, log snapshots
312
313    def _log_incident(self, event: SecurityEvent):
314        """Log incident to SIEM"""
315        logger.info(f"📝 Logging incident: {event.event_id}")
316
317        # Send to SIEM (Splunk, ELK, etc.)