Upload app (1).py

app (1).py

@@ -0,0 +1,380 @@
+import gradio as gr
+import os
+import json
+import datetime
+import pandas as pd
+import matplotlib.pyplot as plt
+import seaborn as sns
+import yaml
+import uuid
+import tempfile
+import shutil
+
+# Demo configuration
+DEMO_CASE_ID = f"DEMO-{uuid.uuid4().hex[:8]}"
+DEMO_OUTPUT_DIR = "demo_output"
+DEMO_EVIDENCE_DIR = os.path.join(DEMO_OUTPUT_DIR, "evidence")
+DEMO_ANALYSIS_DIR = os.path.join(DEMO_OUTPUT_DIR, "analysis")
+DEMO_REPORT_DIR = os.path.join(DEMO_OUTPUT_DIR, "reports")
+
+# Create directories if they don't exist
+os.makedirs(DEMO_EVIDENCE_DIR, exist_ok=True)
+os.makedirs(DEMO_ANALYSIS_DIR, exist_ok=True)
+os.makedirs(DEMO_REPORT_DIR, exist_ok=True)
+
+# Cloud provider connection functions
+def test_aws_connection(access_key, secret_key, region):
+    """Test connection to AWS"""
+    try:
+        import boto3
+        session = boto3.Session(
+            aws_access_key_id=access_key,
+            aws_secret_access_key=secret_key,
+            region_name=region
+        )
+        sts = session.client('sts')
+        identity = sts.get_caller_identity()
+        return True, f"Successfully connected to AWS as {identity['Arn']}"
+    except Exception as e:
+        return False, f"Failed to connect to AWS: {str(e)}"
+
+def test_azure_connection(tenant_id, client_id, client_secret):
+    """Test connection to Azure"""
+    try:
+        from azure.identity import ClientSecretCredential
+        from azure.mgmt.resource import ResourceManagementClient
+        
+        credential = ClientSecretCredential(
+            tenant_id=tenant_id,
+            client_id=client_id,
+            client_secret=client_secret
+        )
+        
+        # Create a resource management client
+        resource_client = ResourceManagementClient(credential, subscription_id)
+        
+        # List resource groups to test the connection
+        resource_groups = list(resource_client.resource_groups.list())
+        return True, f"Successfully connected to Azure. Found {len(resource_groups)} resource groups."
+    except Exception as e:
+        return False, f"Failed to connect to Azure: {str(e)}"
+
+def test_gcp_connection(service_account_json):
+    """Test connection to GCP"""
+    try:
+        import json
+        from google.oauth2 import service_account
+        from google.cloud import storage
+        
+        # Create a temporary file to store the service account JSON
+        fd, path = tempfile.mkstemp()
+        try:
+            with os.fdopen(fd, 'w') as tmp:
+                tmp.write(service_account_json)
+            
+            # Create credentials from the service account file
+            credentials = service_account.Credentials.from_service_account_file(path)
+            
+            # Create a storage client to test the connection
+            storage_client = storage.Client(credentials=credentials)
+            
+            # List buckets to test the connection
+            buckets = list(storage_client.list_buckets())
+            return True, f"Successfully connected to GCP. Found {len(buckets)} storage buckets."
+        finally:
+            os.remove(path)
+    except Exception as e:
+        return False, f"Failed to connect to GCP: {str(e)}"
+
+# Sample data for demonstration
+def generate_sample_data(case_info, cloud_provider, incident_type, use_real_data=False, credentials=None):
+    """Generate sample data for demonstration purposes or collect real data if credentials provided"""
+    
+    if use_real_data and credentials:
+        # This would be where we implement real data collection using the provided credentials
+        # For now, we'll return a message indicating this would use real data
+        return {
+            "timeline": [],
+            "patterns": [],
+            "anomalies": [],
+            "files": {},
+            "message": "In a production deployment, this would collect real data from your cloud provider."
+        }
+    
+    # Create sample timeline data
+    timeline_data = []
+    base_time = datetime.datetime.now() - datetime.timedelta(days=1)
+    
+    # Different events based on incident type
+    if incident_type == "Unauthorized Access":
+        events = [
+            {"event": "Failed login attempt", "source": "Authentication Logs", "severity": "Low"},
+            {"event": "Successful login from unusual IP", "source": "Authentication Logs", "severity": "Medium"},
+            {"event": "User privilege escalation", "source": "IAM Logs", "severity": "High"},
+            {"event": "Access to sensitive data", "source": "Data Access Logs", "severity": "High"},
+            {"event": "Configuration change", "source": "Configuration Logs", "severity": "Medium"},
+            {"event": "New API key created", "source": "IAM Logs", "severity": "High"},
+            {"event": "Data download initiated", "source": "Data Access Logs", "severity": "Critical"},
+            {"event": "Unusual network traffic", "source": "Network Logs", "severity": "Medium"}
+        ]
+    elif incident_type == "Data Exfiltration":
+        events = [
+            {"event": "Large query executed", "source": "Database Logs", "severity": "Medium"},
+            {"event": "Unusual data access pattern", "source": "Data Access Logs", "severity": "Medium"},
+            {"event": "Large data transfer initiated", "source": "Network Logs", "severity": "High"},
+            {"event": "Connection to unknown external endpoint", "source": "Network Logs", "severity": "High"},
+            {"event": "Storage object permissions modified", "source": "Storage Logs", "severity": "Medium"},
+            {"event": "Unusual user behavior", "source": "User Activity Logs", "severity": "Medium"},
+            {"event": "Data archive created", "source": "Storage Logs", "severity": "Medium"},
+            {"event": "Unusual egress traffic spike", "source": "Network Logs", "severity": "Critical"}
+        ]
+    else:  # Ransomware
+        events = [
+            {"event": "Unusual process execution", "source": "System Logs", "severity": "Medium"},
+            {"event": "Multiple file modifications", "source": "File System Logs", "severity": "High"},
+            {"event": "Encryption library loaded", "source": "System Logs", "severity": "High"},
+            {"event": "Mass file type changes", "source": "Storage Logs", "severity": "Critical"},
+            {"event": "Backup deletion attempt", "source": "Backup Logs", "severity": "Critical"},
+            {"event": "Unusual IAM activity", "source": "IAM Logs", "severity": "Medium"},
+            {"event": "Recovery service disabled", "source": "System Logs", "severity": "High"},
+            {"event": "Ransom note created", "source": "File System Logs", "severity": "Critical"}
+        ]
+    
+    # Create timeline with timestamps
+    for i, event in enumerate(events):
+        event_time = base_time + datetime.timedelta(minutes=i*15)
+        timeline_data.append({
+            "timestamp": event_time.isoformat(),
+            "event": event["event"],
+            "source": event["source"],
+            "cloud_provider": cloud_provider,
+            "severity": event["severity"],
+            "case_id": case_info["case_id"]
+        })
+    
+    # Create patterns data
+    patterns = []
+    if incident_type == "Unauthorized Access":
+        patterns = [
+            {"pattern": "Brute Force Login Attempt", "confidence": 0.85, "matched_events": 3},
+            {"pattern": "Privilege Escalation", "confidence": 0.92, "matched_events": 2}
+        ]
+    elif incident_type == "Data Exfiltration":
+        patterns = [
+            {"pattern": "Data Staging Activity", "confidence": 0.88, "matched_events": 3},
+            {"pattern": "Exfiltration Over Alternative Protocol", "confidence": 0.76, "matched_events": 2}
+        ]
+    else:  # Ransomware
+        patterns = [
+            {"pattern": "Mass File Encryption", "confidence": 0.94, "matched_events": 4},
+            {"pattern": "Defense Evasion", "confidence": 0.81, "matched_events": 3}
+        ]
+    
+    # Create anomalies data
+    anomalies = []
+    if incident_type == "Unauthorized Access":
+        anomalies = [
+            {"anomaly": "Login from unusual location", "deviation": 3.6, "severity": "High"},
+            {"anomaly": "Off-hours access", "deviation": 2.8, "severity": "Medium"}
+        ]
+    elif incident_type == "Data Exfiltration":
+        anomalies = [
+            {"anomaly": "Unusual data access volume", "deviation": 4.2, "severity": "High"},
+            {"anomaly": "Abnormal query pattern", "deviation": 3.1, "severity": "Medium"}
+        ]
+    else:  # Ransomware
+        anomalies = [
+            {"anomaly": "Unusual file system activity", "deviation": 4.7, "severity": "Critical"},
+            {"anomaly": "Suspicious process behavior", "deviation": 3.9, "severity": "High"}
+        ]
+    
+    # Save data to files
+    timeline_file = os.path.join(DEMO_EVIDENCE_DIR, f"{DEMO_CASE_ID}_timeline.json")
+    patterns_file = os.path.join(DEMO_ANALYSIS_DIR, f"{DEMO_CASE_ID}_patterns.json")
+    anomalies_file = os.path.join(DEMO_ANALYSIS_DIR, f"{DEMO_CASE_ID}_anomalies.json")
+    
+    with open(timeline_file, 'w') as f:
+        json.dump(timeline_data, f, indent=2)
+    
+    with open(patterns_file, 'w') as f:
+        json.dump(patterns, f, indent=2)
+    
+    with open(anomalies_file, 'w') as f:
+        json.dump(anomalies, f, indent=2)
+    
+    return {
+        "timeline": timeline_data,
+        "patterns": patterns,
+        "anomalies": anomalies,
+        "files": {
+            "timeline": timeline_file,
+            "patterns": patterns_file,
+            "anomalies": anomalies_file
+        }
+    }
+
+def analyze_evidence(data):
+    """Perform analysis on the evidence data"""
+    
+    # If there's no timeline data, return empty results
+    if not data["timeline"]:
+        return {
+            "severity_counts": {},
+            "source_counts": {},
+            "charts": {
+                "analysis": None,
+                "timeline": None
+            }
+        }
+    
+    # Convert timeline to DataFrame for analysis
+    timeline_df = pd.DataFrame(data["timeline"])
+    timeline_df["timestamp"] = pd.to_datetime(timeline_df["timestamp"])
+    
+    # Sort by timestamp
+    timeline_df = timeline_df.sort_values("timestamp")
+    
+    # Count events by severity
+    severity_counts = timeline_df["severity"].value_counts()
+    
+    # Count events by source
+    source_counts = timeline_df["source"].value_counts()
+    
+    # Create visualizations
+    fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(12, 5))
+    
+    # Severity pie chart
+    ax1.pie(severity_counts, labels=severity_counts.index, autopct='%1.1f%%', 
+            colors=sns.color_palette("YlOrRd", len(severity_counts)))
+    ax1.set_title("Events by Severity")
+    
+    # Source bar chart
+    sns.barplot(x=source_counts.values, y=source_counts.index, ax=ax2, palette="viridis")
+    ax2.set_title("Events by Source")
+    ax2.set_xlabel("Count")
+    
+    # Save the figure
+    chart_file = os.path.join(DEMO_ANALYSIS_DIR, f"{DEMO_CASE_ID}_analysis_charts.png")
+    plt.tight_layout()
+    plt.savefig(chart_file)
+    plt.close()
+    
+    # Create a timeline visualization
+    plt.figure(figsize=(12, 6))
+    
+    # Create a categorical y-axis based on source
+    sources = timeline_df["source"].unique()
+    source_map = {source: i for i, source in enumerate(sources)}
+    timeline_df["source_num"] = timeline_df["source"].map(source_map)
+    
+    # Map severity to color
+    severity_colors = {
+        "Low": "green",
+        "Medium": "blue",
+        "High": "orange",
+        "Critical": "red"
+    }
+    colors = timeline_df["severity"].map(severity_colors)
+    
+    # Plot the timeline
+    plt.scatter(timeline_df["timestamp"], timeline_df["source_num"], c=colors, s=100)
+    
+    # Add event labels
+    for i, row in timeline_df.iterrows():
+        plt.text(row["timestamp"], row["source_num"], row["event"], 
+                 fontsize=8, ha="right", va="bottom", rotation=25)
+    
+    plt.yticks(range(len(sources)), sources)
+    plt.xlabel("Time")
+    plt.ylabel("Event Source")
+    plt.title("Incident Timeline")
+    
+    # Save the timeline
+    timeline_chart = os.path.join(DEMO_ANALYSIS_DIR, f"{DEMO_CASE_ID}_timeline_chart.png")
+    plt.tight_layout()
+    plt.savefig(timeline_chart)
+    plt.close()
+    
+    return {
+        "severity_counts": severity_counts.to_dict(),
+        "source_counts": source_counts.to_dict(),
+        "charts": {
+            "analysis": chart_file,
+            "timeline": timeline_chart
+        }
+    }
+
+def generate_report(case_info, data, analysis, report_format):
+    """Generate a report based on the analysis"""
+    
+    # Create report content
+    report = {
+        "case_information": case_info,
+        "executive_summary": f"This report presents the findings of a forensic investigation into a {case_info['incident_type']} incident in {case_info['cloud_provider']} cloud environment.",
+        "timeline": data["timeline"],
+        "patterns_detected": data["patterns"],
+        "anomalies_detected": data["anomalies"],
+        "analysis_results": {
+            "severity_distribution": analysis.get("severity_counts", {}),
+            "source_distribution": analysis.get("source_counts", {})
+        },
+        "recommendations": [
+            "Implement multi-factor authentication for all privileged accounts",
+            "Review and restrict IAM permissions following principle of least privilege",
+            "Enable comprehensive logging across all cloud services",
+            "Implement automated alerting for suspicious activities",
+            "Conduct regular security assessments of cloud environments"
+        ]
+    }
+    
+    # Save report in requested format
+    if report_format == "JSON":
+        report_file = os.path.join(DEMO_REPORT_DIR, f"{DEMO_CASE_ID}_report.json")
+        with open(report_file, 'w') as f:
+            json.dump(report, f, indent=2)
+    else:  # HTML
+        # Create a simple HTML report
+        html_content = f"""
+        <!DOCTYPE html>
+        <html>
+        <head>
+            <title>Forensic Analysis Report - {case_info['case_id']}</title>
+            <style>
+                body {{ font-family: Arial, sans-serif; margin: 40px; }}
+                h1, h2, h3 {{ color: #2c3e50; }}
+                .section {{ margin-bottom: 30px; }}
+                .severity-high {{ color: #e74c3c; }}
+                .severity-medium {{ color: #f39c12; }}
+                .severity-low {{ color: #27ae60; }}
+                table {{ border-collapse: collapse; width: 100%; }}
+                th, td {{ border: 1px solid #ddd; padding: 8px; text-align: left; }}
+                th {{ background-color: #f2f2f2; }}
+                tr:nth-child(even) {{ background-color: #f9f9f9; }}
+                .chart-container {{ display: flex; justify-content: center; margin: 20px 0; }}
+                .chart {{ max-width: 100%; height: auto; margin: 10px; }}
+                .message {{ background-color: #f8f9fa; padding: 15px; border-left: 5px solid #4e73df; margin-bottom: 20px; }}
+            </style>
+        </head>
+        <body>
+            <h1>Cloud Forensics Analysis Report</h1>
+            
+            <div class="section">
+                <h2>Case Information</h2>
+                <p><strong>Case ID:</strong> {case_info['case_id']}</p>
+                <p><strong>Investigator:</strong> {case_info['investigator']}</p>
+                <p><strong>Organization:</strong> {case_info['organization']}</p>
+                <p><strong>Cloud Provider:</strong> {case_info['cloud_provider']}</p>
+                <p><strong>Incident Type:</strong> {case_info['incident_type']}</p>
+                <p><strong>Report Date:</strong> {datetime.datetime.now().strftime('%Y-%m-%d')}</p>
+            </div>
+            
+            <div class="section">
+                <h2>Executive Summary</h2>
+                <p>{report['executive_summary']}</p>
+        """
+        
+        # Add message if using real data
+        if "message" in data:
+            html_content += f"""
+                <div class="mes
+(Content truncated due to size limit. Use line ranges to read in chunks)