How to Use AI to Build Defensive Cybersecurity Tools (Save 80% Development Time)

#!/usr/bin/env python3
"""
Defensive Network Traffic Analyzer
Purpose: Monitor YOUR network for unusual patterns
Author: AI-Generated, modified for production use
"""

from scapy.all import sniff, IP, TCP, UDP
import json
import datetime
import logging
from collections import defaultdict, Counter

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    filename='network_monitor.log'
)

class NetworkMonitor:
    def __init__(self, interface="eth0"):
        self.interface = interface
        self.connection_counts = defaultdict(int)
        self.port_activity = Counter()
        self.suspicious_ips = set()
        
        # Define suspicious ports (commonly targeted)
        self.suspicious_ports = {22, 23, 135, 139, 445, 1433, 3389, 5432}
        
    def analyze_packet(self, packet):
        """Analyze individual packet for suspicious patterns"""
        if IP in packet:
            src_ip = packet[IP].src
            dst_ip = packet[IP].dst
            
            # Skip local loopback
            if src_ip.startswith('127.') or dst_ip.startswith('127.'):
                return
            
            # Track connection counts
            connection_key = f"{src_ip}->{dst_ip}"
            self.connection_counts[connection_key] += 1
            
            # Analyze port activity
            if TCP in packet:
                dst_port = packet[TCP].dport
                self.port_activity[dst_port] += 1
                
                # Flag suspicious port activity
                if dst_port in self.suspicious_ports:
                    self.flag_suspicious_activity(src_ip, dst_port, "TCP")
            
            elif UDP in packet:
                dst_port = packet[UDP].dport
                self.port_activity[dst_port] += 1
    
    def flag_suspicious_activity(self, ip, port, protocol):
        """Log suspicious activity for review"""
        alert = {
            'timestamp': datetime.datetime.now().isoformat(),
            'source_ip': ip,
            'target_port': port,
            'protocol': protocol,
            'alert_type': 'suspicious_port_activity'
        }
        
        logging.warning(f"Suspicious activity: {json.dumps(alert)}")
        print(f"⚠️  Alert: {ip} accessing port {port}/{protocol}")
    
    def generate_report(self):
        """Generate summary report of network activity"""
        print("\n🔍 Network Activity Report")
        print("=" * 50)
        
        # Top connections
        print("\nTop 10 Connection Patterns:")
        for conn, count in self.connection_counts.most_common(10):
            print(f"  {conn}: {count} packets")
        
        # Port activity summary  
        print("\nTop 10 Port Activity:")
        for port, count in self.port_activity.most_common(10):
            status = "⚠️ " if port in self.suspicious_ports else "✅"
            print(f"  {status} Port {port}: {count} connections")
    
    def start_monitoring(self, packet_count=1000):
        """Start packet capture and analysis"""
        print(f"🔍 Starting network monitoring on {self.interface}")
        print(f"📊 Will analyze {packet_count} packets")
        print("🛑 Press Ctrl+C to stop early and generate report")
        
        try:
            sniff(
                iface=self.interface, 
                prn=self.analyze_packet, 
                count=packet_count,
                store=0  # Don't store packets in memory
            )
        except KeyboardInterrupt:
            print("\n🛑 Monitoring stopped by user")
        
        self.generate_report()

if __name__ == "__main__":
    # Initialize monitor
    monitor = NetworkMonitor(interface="eth0")  # Change to your interface
    
    # Start monitoring
    monitor.start_monitoring(packet_count=5000)

class SmartFilter:
    def __init__(self):
        self.baseline_established = False
        self.normal_patterns = {
            'common_ports': set(),
            'regular_connections': set(),
            'peak_hours': set()
        }
    
    def learn_baseline(self, duration_minutes=60):
        """AI-suggested approach: Learn normal network patterns"""
        print(f"🧠 Learning baseline network behavior for {duration_minutes} minutes")
        
        # This would collect normal traffic patterns
        # Implementation details generated by AI
        baseline_data = self.collect_baseline_traffic(duration_minutes)
        self.establish_normal_patterns(baseline_data)
        self.baseline_established = True
    
    def is_anomalous(self, connection_pattern):
        """Determine if traffic pattern is anomalous"""
        if not self.baseline_established:
            return False  # Don't alert during learning phase
        
        # AI-generated anomaly detection logic
        return self.detect_deviation_from_baseline(connection_pattern)

#!/usr/bin/env python3
"""
Intelligent Security Log Analyzer
Purpose: Find real threats in massive log files
"""

import re
import json
import datetime
from collections import defaultdict, Counter
from typing import Dict, List, Any

class SecurityLogAnalyzer:
    def __init__(self):
        self.threat_patterns = self._load_threat_patterns()
        self.false_positive_filters = self._load_fp_filters()
        self.incident_counter = Counter()
        
    def _load_threat_patterns(self) -> Dict[str, Dict]:
        """AI-generated threat detection patterns"""
        return {
            'brute_force': {
                'pattern': r'Failed password for .* from ([\d\.]+)',
                'threshold': 5,  # 5+ failures in timeframe
                'timeframe': 300,  # 5 minutes
                'severity': 'HIGH'
            },
            'privilege_escalation': {
                'pattern': r'sudo.*COMMAND=.*(/bin/bash|/bin/sh)',
                'threshold': 1,
                'timeframe': 60,
                'severity': 'CRITICAL'
            },
            'unusual_file_access': {
                'pattern': r'DENIED.*(/etc/passwd|/etc/shadow|\.ssh/)',
                'threshold': 1,
                'timeframe': 60,
                'severity': 'HIGH'
            }
        }
    
    def _load_fp_filters(self) -> List[str]:
        """Common false positive patterns to ignore"""
        return [
            r'.*monitoring\.company\.com.*',  # Internal monitoring
            r'.*backup.*service.*',           # Backup services
            r'.*cron.*daily.*',              # Scheduled tasks
        ]
    
    def analyze_log_file(self, filepath: str) -> List[Dict[str, Any]]:
        """Analyze security log file for threats"""
        incidents = []
        
        print(f"🔍 Analyzing {filepath}")
        
        with open(filepath, 'r') as f:
            logs = f.readlines()
        
        print(f"📄 Processing {len(logs)} log entries")
        
        for log_entry in logs:
            # Check against each threat pattern
            for threat_name, config in self.threat_patterns.items():
                match = re.search(config['pattern'], log_entry)
                
                if match and not self._is_false_positive(log_entry):
                    incident = {
                        'timestamp': self._extract_timestamp(log_entry),
                        'threat_type': threat_name,
                        'severity': config['severity'],
                        'matched_content': match.group(0),
                        'source_ip': match.group(1) if match.groups() else 'unknown',
                        'raw_log': log_entry.strip()
                    }
                    
                    incidents.append(incident)
                    self.incident_counter[threat_name] += 1
        
        return incidents
    
    def _is_false_positive(self, log_entry: str) -> bool:
        """Check if log entry matches known false positive patterns"""
        for fp_pattern in self.false_positive_filters:
            if re.search(fp_pattern, log_entry):
                return True
        return False
    
    def _extract_timestamp(self, log_entry: str) -> str:
        """Extract timestamp from log entry"""
        # Simplified timestamp extraction
        timestamp_pattern = r'(\w{3}\s+\d{1,2}\s+\d{2}:\d{2}:\d{2})'
        match = re.search(timestamp_pattern, log_entry)
        return match.group(1) if match else "unknown"
    
    def generate_security_report(self, incidents: List[Dict]) -> None:
        """Generate human-readable security report"""
        print("\n🚨 Security Incident Report")
        print("=" * 60)
        
        if not incidents:
            print("✅ No security incidents detected")
            return
        
        # Group by severity
        critical = [i for i in incidents if i['severity'] == 'CRITICAL']
        high = [i for i in incidents if i['severity'] == 'HIGH']
        
        print(f"\n🔴 Critical Incidents: {len(critical)}")
        for incident in critical[:5]:  # Show top 5
            print(f"  • {incident['threat_type']}: {incident['source_ip']}")
        
        print(f"\n🟡 High Priority Incidents: {len(high)}")
        for incident in high[:5]:  # Show top 5
            print(f"  • {incident['threat_type']}: {incident['source_ip']}")
        
        print(f"\n📊 Incident Summary:")
        for threat, count in self.incident_counter.most_common():
            print(f"  {threat}: {count} incidents")
    
    def export_incidents(self, incidents: List[Dict], output_file: str) -> None:
        """Export incidents to JSON for SIEM integration"""
        with open(output_file, 'w') as f:
            json.dump(incidents, f, indent=2, default=str)
        
        print(f"📁 Exported {len(incidents)} incidents to {output_file}")

# Example usage
if __name__ == "__main__":
    analyzer = SecurityLogAnalyzer()
    
    # Analyze your security logs
    incidents = analyzer.analyze_log_file('/var/log/auth.log')
    
    # Generate report
    analyzer.generate_security_report(incidents)
    
    # Export for further analysis
    analyzer.export_incidents(incidents, 'security_incidents.json')

#!/usr/bin/env python3
"""
Defensive Vulnerability Assessment Tool
Purpose: Identify security weaknesses in YOUR infrastructure
WARNING: Only use on systems you own or have written permission to test
"""

import socket
import requests
import subprocess
import json
import datetime
from urllib.parse import urljoin
import ssl
from typing import List, Dict, Any

class VulnerabilityScanner:
    def __init__(self, target_host: str):
        self.target = target_host
        self.findings = []
        self.scan_timestamp = datetime.datetime.now()
        
        print(f"🔍 Initializing vulnerability scan for {target_host}")
        print("⚠️  Only scanning systems you own or have permission to test")
    
    def check_open_ports(self, port_list: List[int] = None) -> None:
        """Check for unnecessarily open ports"""
        if port_list is None:
            # Common ports that shouldn't be open to internet
            port_list = [21, 22, 23, 25, 53, 135, 139, 445, 1433, 3389, 5432]
        
        print("🔍 Checking for open ports...")
        open_ports = []
        
        for port in port_list:
            try:
                sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
                sock.settimeout(3)
                result = sock.connect_ex((self.target, port))
                
                if result == 0:  # Port is open
                    open_ports.append(port)
                    self.findings.append({
                        'category': 'Network Security',
                        'severity': 'MEDIUM',
                        'issue': f'Open port {port}',
                        'description': f'Port {port} is accessible from external networks',
                        'recommendation': f'Close port {port} if not required or restrict access with firewall rules'
                    })
                
                sock.close()
            except socket.gaierror:
                print(f"❌ Could not resolve hostname: {self.target}")
                return
        
        print(f"🔍 Found {len(open_ports)} open ports: {open_ports}")
    
    def check_ssl_configuration(self, port: int = 443) -> None:
        """Check SSL/TLS configuration"""
        print("🔒 Checking SSL/TLS configuration...")
        
        try:
            context = ssl.create_default_context()
            with socket.create_connection((self.target, port), timeout=10) as sock:
                with context.wrap_socket(sock, server_hostname=self.target) as ssock:
                    cert = ssock.getpeercert()
                    protocol = ssock.version()
                    
                    # Check certificate expiration
                    not_after = datetime.datetime.strptime(
                        cert['notAfter'], '%b %d %H:%M:%S %Y %Z'
                    )
                    days_until_expiry = (not_after - datetime.datetime.now()).days
                    
                    if days_until_expiry < 30:
                        self.findings.append({
                            'category': 'SSL/TLS Security',
                            'severity': 'HIGH' if days_until_expiry < 7 else 'MEDIUM',
                            'issue': 'Certificate expiring soon',
                            'description': f'SSL certificate expires in {days_until_expiry} days',
                            'recommendation': 'Renew SSL certificate before expiration'
                        })
                    
                    # Check for weak protocols
                    if protocol in ['TLSv1', 'TLSv1.1']:
                        self.findings.append({
                            'category': 'SSL/TLS Security',
                            'severity': 'HIGH',
                            'issue': 'Weak TLS protocol',
                            'description': f'Server supports weak protocol: {protocol}',
                            'recommendation': 'Disable TLS 1.0 and 1.1, use TLS 1.2+ only'
                        })
                    
                    print(f"✅ SSL check complete - Protocol: {protocol}")
        
        except Exception as e:
            print(f"❌ SSL check failed: {e}")
    
    def check_web_application_security(self, base_url: str) -> None:
        """Basic web application security checks"""
        print("🌐 Checking web application security...")
        
        try:
            # Check for security headers
            response = requests.get(base_url, timeout=10)
            headers = response.headers
            
            # Important security headers
            security_headers = {
                'X-Content-Type-Options': 'nosniff',
                'X-Frame-Options': ['DENY', 'SAMEORIGIN'],
                'X-XSS-Protection': '1; mode=block',
                'Strict-Transport-Security': 'max-age=',
                'Content-Security-Policy': 'default-src'
            }
            
            for header, expected in security_headers.items():
                if header not in headers:
                    self.findings.append({
                        'category': 'Web Application Security',
                        'severity': 'MEDIUM',
                        'issue': f'Missing security header: {header}',
                        'description': f'Application does not set {header} header',
                        'recommendation': f'Add {header} header to prevent common attacks'
                    })
                elif isinstance(expected, list):
                    if not any(exp in headers[header] for exp in expected):
                        self.findings.append({
                            'category': 'Web Application Security',
                            'severity': 'LOW',
                            'issue': f'Weak {header} header',
                            'description': f'{header} header may not provide adequate protection',
                            'recommendation': f'Review and strengthen {header} configuration'
                        })
            
            print(f"🌐 Web security check complete")
            
        except Exception as e:
            print(f"❌ Web application check failed: {e}")
    
    def check_common_vulnerabilities(self) -> None:
        """Check for common configuration vulnerabilities"""
        print("🔍 Checking for common vulnerabilities...")
        
        # Check if SSH is running with default settings
        try:
            sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            sock.settimeout(5)
            result = sock.connect_ex((self.target, 22))
            
            if result == 0:
                # SSH is open - check banner for version info
                sock.send(b"SSH-2.0-Scanner\r\n")
                banner = sock.recv(1024).decode().strip()
                
                if 'OpenSSH' in banner:
                    # Extract version (simplified)
                    if any(weak in banner for weak in ['OpenSSH_7.', 'OpenSSH_6.', 'OpenSSH_5.']):
                        self.findings.append({
                            'category': 'Software Vulnerabilities',
                            'severity': 'HIGH',
                            'issue': 'Outdated SSH server',
                            'description': f'SSH server version may have known vulnerabilities: {banner}',
                            'recommendation': 'Update SSH server to latest version'
                        })
            
            sock.close()
            
        except Exception as e:
            pass  # SSH not available or accessible
    
    def generate_report(self) -> None:
        """Generate comprehensive vulnerability report"""
        print("\n" + "="*80)
        print("🛡️  VULNERABILITY ASSESSMENT REPORT")
        print("="*80)
        print(f"Target: {self.target}")
        print(f"Scan Date: {self.scan_timestamp.strftime('%Y-%m-%d %H:%M:%S')}")
        print(f"Total Findings: {len(self.findings)}")
        
        if not self.findings:
            print("\n✅ No vulnerabilities detected!")
            return
        
        # Group by severity
        critical = [f for f in self.findings if f['severity'] == 'CRITICAL']
        high = [f for f in self.findings if f['severity'] == 'HIGH']
        medium = [f for f in self.findings if f['severity'] == 'MEDIUM']
        low = [f for f in self.findings if f['severity'] == 'LOW']
        
        def print_findings(findings, title):
            if findings:
                print(f"\n{title} ({len(findings)} findings):")
                print("-" * 40)
                for i, finding in enumerate(findings, 1):
                    print(f"{i}. {finding['issue']}")
                    print(f"   Category: {finding['category']}")
                    print(f"   Description: {finding['description']}")
                    print(f"   Fix: {finding['recommendation']}")
                    print()
        
        print_findings(critical, "🔴 CRITICAL ISSUES")
        print_findings(high, "🟡 HIGH PRIORITY ISSUES")
        print_findings(medium, "🟠 MEDIUM PRIORITY ISSUES")
        print_findings(low, "🟢 LOW PRIORITY ISSUES")
        
        # Summary recommendations
        print("\n📋 IMMEDIATE ACTION ITEMS:")
        print("-" * 40)
        priority_actions = critical + high
        for i, action in enumerate(priority_actions[:5], 1):
            print(f"{i}. {action['recommendation']}")
    
    def export_report(self, filename: str = None) -> None:
        """Export detailed report to JSON"""
        if filename is None:
            filename = f"vuln_report_{self.target}_{datetime.datetime.now().strftime('%Y%m%d_%H%M%S')}.json"
        
        report = {
            'target': self.target,
            'scan_timestamp': self.scan_timestamp.isoformat(),
            'findings': self.findings,
            'summary': {
                'total_findings': len(self.findings),
                'critical': len([f for f in self.findings if f['severity'] == 'CRITICAL']),
                'high': len([f for f in self.findings if f['severity'] == 'HIGH']),
                'medium': len([f for f in self.findings if f['severity'] == 'MEDIUM']),
                'low': len([f for f in self.findings if f['severity'] == 'LOW'])
            }
        }
        
        with open(filename, 'w') as f:
            json.dump(report, f, indent=2, default=str)
        
        print(f"\n📄 Detailed report exported to: {filename}")

# Example usage
if __name__ == "__main__":
    # Only scan systems you own!
    target_system = input("Enter target system (you must own this): ")
    
    if not target_system:
        print("❌ No target specified")
        exit(1)
    
    # Confirm ownership
    confirm = input(f"⚠️  Confirm you own/have permission to scan {target_system} (yes/no): ")
    if confirm.lower() != 'yes':
        print("❌ Scan cancelled - only scan systems you own")
        exit(1)
    
    # Initialize scanner
    scanner = VulnerabilityScanner(target_system)
    
    # Run scans
    scanner.check_open_ports()
    scanner.check_ssl_configuration()
    scanner.check_web_application_security(f"https://{target_system}")
    scanner.check_common_vulnerabilities()
    
    # Generate reports
    scanner.generate_report()
    scanner.export_report()