Cybersecurity Services

# Automated Security Scan
import nmap
import requests
from bs4 import BeautifulSoup

class SecurityScanner:
    def __init__(self, target):
        self.target = target
        self.nm = nmap.PortScanner()
        
    def port_scan(self):
        self.nm.scan(self.target, '1-1000')
        open_ports = []
        
        for host in self.nm.all_hosts():
            for proto in self.nm[host].all_protocols():
                ports = self.nm[host][proto].keys()
                for port in ports:
                    state = self.nm[host][proto][port]['state']
                    if state == 'open':
                        open_ports.append(port)
        return open_ports
    
    def web_vulnerability_scan(self, url):
        vulnerabilities = []
        
        # Check for SQL injection
        sql_payloads = ["'", "1' OR '1'='1", "'; DROP TABLE users; --"]
        for payload in sql_payloads:
            test_url = f"{url}?id={payload}"
            response = requests.get(test_url)
            if "error" in response.text.lower():
                vulnerabilities.append(f"Potential SQL injection: {payload}")
        
        return vulnerabilities

# Vulnerability Assessment Framework
import json
import subprocess
from datetime import datetime

class VulnerabilityAssessment:
    def __init__(self):
        self.findings = []
        self.severity_levels = ['Critical', 'High', 'Medium', 'Low', 'Info']
    
    def scan_dependencies(self, project_path):
        # Check for known vulnerable dependencies
        result = subprocess.run(['npm', 'audit', '--json'], 
                              cwd=project_path, capture_output=True, text=True)
        
        if result.stdout:
            audit_data = json.loads(result.stdout)
            for vuln in audit_data.get('vulnerabilities', {}):
                self.findings.append({
                    'type': 'Dependency Vulnerability',
                    'severity': vuln.get('severity', 'Unknown'),
                    'package': vuln.get('name'),
                    'description': vuln.get('title'),
                    'fix': vuln.get('fixAvailable', False)
                })
    
    def check_security_headers(self, url):
        import requests
        response = requests.get(url)
        headers = response.headers
        
        security_headers = {
            'X-Frame-Options': 'Clickjacking protection',
            'X-XSS-Protection': 'XSS protection',
            'X-Content-Type-Options': 'MIME type sniffing protection',
            'Strict-Transport-Security': 'HTTPS enforcement'
        }
        
        for header, description in security_headers.items():
            if header not in headers:
                self.findings.append({
                    'type': 'Missing Security Header',
                    'severity': 'Medium',
                    'header': header,
                    'description': description
                })
    
    def generate_report(self):
        return {
            'scan_date': datetime.now().isoformat(),
            'total_findings': len(self.findings),
            'findings_by_severity': self._group_by_severity(),
            'detailed_findings': self.findings
        }

# Infrastructure Security Automation
import boto3
import json
from datetime import datetime

class AWSSecurityHardening:
    def __init__(self):
        self.ec2 = boto3.client('ec2')
        self.iam = boto3.client('iam')
        self.cloudtrail = boto3.client('cloudtrail')
    
    def audit_security_groups(self):
        security_issues = []
        response = self.ec2.describe_security_groups()
        
        for sg in response['SecurityGroups']:
            for rule in sg['IpPermissions']:
                for ip_range in rule.get('IpRanges', []):
                    if ip_range.get('CidrIp') == '0.0.0.0/0':
                        security_issues.append({
                            'type': 'Open Security Group',
                            'group_id': sg['GroupId'],
                            'port': rule.get('FromPort'),
                            'protocol': rule.get('IpProtocol'),
                            'risk': 'High'
                        })
        return security_issues
    
    def enable_cloudtrail(self, trail_name, s3_bucket):
        try:
            self.cloudtrail.create_trail(
                Name=trail_name,
                S3BucketName=s3_bucket,
                IncludeGlobalServiceEvents=True,
                IsMultiRegionTrail=True,
                EnableLogFileValidation=True
            )
            
            self.cloudtrail.start_logging(Name=trail_name)
            return {"status": "success", "message": "CloudTrail enabled"}
        except Exception as e:
            return {"status": "error", "message": str(e)}
    
    def check_iam_policies(self):
        risky_policies = []
        response = self.iam.list_policies(Scope='Local')
        
        for policy in response['Policies']:
            policy_version = self.iam.get_policy_version(
                PolicyArn=policy['Arn'],
                VersionId=policy['DefaultVersionId']
            )
            
            document = policy_version['PolicyVersion']['Document']
            if self._has_admin_access(document):
                risky_policies.append({
                    'policy_name': policy['PolicyName'],
                    'risk': 'Administrative access detected'
                })
        
        return risky_policies

# API Security Implementation
import jwt
import hashlib
import time
from functools import wraps
from flask import Flask, request, jsonify

app = Flask(__name__)

class APISecurityManager:
    def __init__(self, secret_key):
        self.secret_key = secret_key
        self.rate_limits = {}
    
    def generate_token(self, user_id, permissions):
        payload = {
            'user_id': user_id,
            'permissions': permissions,
            'exp': time.time() + 3600,  # 1 hour expiration
            'iat': time.time()
        }
        return jwt.encode(payload, self.secret_key, algorithm='HS256')
    
    def verify_token(self, token):
        try:
            payload = jwt.decode(token, self.secret_key, algorithms=['HS256'])
            return payload
        except jwt.ExpiredSignatureError:
            return None
        except jwt.InvalidTokenError:
            return None
    
    def rate_limit(self, max_requests=100, window=3600):
        def decorator(f):
            @wraps(f)
            def decorated_function(*args, **kwargs):
                client_ip = request.remote_addr
                current_time = time.time()
                
                if client_ip not in self.rate_limits:
                    self.rate_limits[client_ip] = []
                
                # Clean old requests
                self.rate_limits[client_ip] = [
                    req_time for req_time in self.rate_limits[client_ip]
                    if current_time - req_time < window
                ]
                
                if len(self.rate_limits[client_ip]) >= max_requests:
                    return jsonify({'error': 'Rate limit exceeded'}), 429
                
                self.rate_limits[client_ip].append(current_time)
                return f(*args, **kwargs)
            return decorated_function
        return decorator
    
    def validate_input(self, data, schema):
        # Input validation against schema
        for field, rules in schema.items():
            if field not in data and rules.get('required', False):
                return False, f"Missing required field: {field}"
            
            if field in data:
                value = data[field]
                if 'type' in rules and not isinstance(value, rules['type']):
                    return False, f"Invalid type for {field}"
                
                if 'max_length' in rules and len(str(value)) > rules['max_length']:
                    return False, f"Field {field} exceeds maximum length"
        
        return True, "Valid"

# Security Monitoring System
import json
import logging
from datetime import datetime, timedelta
import smtplib
from email.mime.text import MIMEText

class SecurityMonitor:
    def __init__(self):
        self.alerts = []
        self.threat_indicators = []
        self.setup_logging()
    
    def setup_logging(self):
        logging.basicConfig(
            level=logging.INFO,
            format='%(asctime)s - %(levelname)s - %(message)s',
            handlers=[
                logging.FileHandler('security.log'),
                logging.StreamHandler()
            ]
        )
        self.logger = logging.getLogger(__name__)
    
    def analyze_log_entry(self, log_entry):
        suspicious_patterns = [
            'failed login',
            'privilege escalation',
            'unusual network activity',
            'malware detected',
            'unauthorized access'
        ]
        
        for pattern in suspicious_patterns:
            if pattern in log_entry.lower():
                alert = {
                    'timestamp': datetime.now().isoformat(),
                    'severity': self._calculate_severity(pattern),
                    'pattern': pattern,
                    'log_entry': log_entry,
                    'source_ip': self._extract_ip(log_entry)
                }
                self.alerts.append(alert)
                self._send_alert(alert)
    
    def _calculate_severity(self, pattern):
        high_severity = ['privilege escalation', 'malware detected']
        medium_severity = ['failed login', 'unauthorized access']
        
        if pattern in high_severity:
            return 'HIGH'
        elif pattern in medium_severity:
            return 'MEDIUM'
        else:
            return 'LOW'
    
    def generate_security_report(self):
        last_24h = datetime.now() - timedelta(hours=24)
        recent_alerts = [
            alert for alert in self.alerts
            if datetime.fromisoformat(alert['timestamp']) > last_24h
        ]
        
        return {
            'report_date': datetime.now().isoformat(),
            'total_alerts_24h': len(recent_alerts),
            'severity_breakdown': self._group_by_severity(recent_alerts),
            'top_threats': self._identify_top_threats(recent_alerts),
            'recommendations': self._generate_recommendations(recent_alerts)
        }