start_app.py

#!/usr/bin/env python3
"""
VDF iOS Forensics Application - Simple Development Server

A simplified launcher for development and testing of the VDF iOS Forensics toolkit.
This creates a minimal web interface for testing the forensic extraction capabilities.
"""

import os
import sys
import logging
import asyncio
import json
from pathlib import Path
from datetime import datetime
from dataclasses import dataclass, asdict
from typing import Dict, List, Optional
from enum import Enum

# Add the current directory to Python path
sys.path.insert(0, str(Path(__file__).parent))

# Simple HTTP server for development
from http.server import HTTPServer, SimpleHTTPRequestHandler
import urllib.parse
import threading
import webbrowser

# Configuration
SERVER_HOST = "localhost"
SERVER_PORT = 8080
STATIC_DIR = Path(__file__).parent / "app" / "static"


class SimpleForensicsHandler(SimpleHTTPRequestHandler):
    """Simple HTTP handler for the forensics application."""
    
    # Class-level storage for jobs (shared across requests)
    jobs = {}
    
    def __init__(self, *args, **kwargs):
        super().__init__(*args, directory=str(STATIC_DIR), **kwargs)
    
    def do_GET(self):
        """Handle GET requests."""
        # Handle API requests first
        if self.path.startswith("/api/"):
            self.handle_api_request()
        elif self.path == "/" or self.path == "/app":
            # Serve the main application
            self.path = "/index.html"
            return super().do_GET()
        else:
            return super().do_GET()
    
    def do_POST(self):
        """Handle POST requests."""
        if self.path.startswith("/api/"):
            self.handle_api_request()
        else:
            self.send_error(404)
    
    def do_OPTIONS(self):
        """Handle OPTIONS requests for CORS."""
        self.send_response(200)
        self.send_header("Access-Control-Allow-Origin", "*")
        self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS")
        self.send_header("Access-Control-Allow-Headers", "Content-Type")
        self.end_headers()
    
    def handle_api_request(self):
        """Handle API requests with mock responses."""
        try:
            print(f"API Request: {self.command} {self.path}")  # Debug logging
            
            if self.path == "/api/analysis/start":
                # Read POST data if present
                content_length = int(self.headers.get('Content-Length', 0))
                if content_length > 0:
                    post_data = self.rfile.read(content_length)
                    try:
                        request_data = json.loads(post_data.decode('utf-8'))
                        print(f"Request data: {request_data}")
                    except:
                        request_data = {}
                
                # Mock job creation
                job_id = f"job_{datetime.now().strftime('%Y%m%d_%H%M%S')}"
                SimpleForensicsHandler.jobs[job_id] = {
                    "job_id": job_id,
                    "status": "pending",
                    "progress": 0.0,
                    "results_count": 0,
                    "threats_detected": 0,
                    "created_at": datetime.now().isoformat(),
                    "error_message": None
                }
                
                # Start mock progress
                threading.Thread(target=self.simulate_job_progress, args=(job_id,)).start()
                
                response = {"job_id": job_id, "message": "Analysis started successfully"}
                self.send_json_response(response)
                
            elif self.path.startswith("/api/jobs/") and self.path.endswith("/status"):
                # Extract job ID
                path_parts = self.path.split("/")
                if len(path_parts) >= 4:
                    job_id = path_parts[3]
                    if job_id in SimpleForensicsHandler.jobs:
                        self.send_json_response(SimpleForensicsHandler.jobs[job_id])
                    else:
                        self.send_json_error(404, "Job not found")
                else:
                    self.send_json_error(400, "Invalid job ID")
                    
            elif self.path == "/api/jobs":
                # List all jobs
                self.send_json_response(list(SimpleForensicsHandler.jobs.values()))
                
            elif self.path.startswith("/api/jobs/") and self.path.endswith("/results"):
                # Mock results
                path_parts = self.path.split("/")
                if len(path_parts) >= 4:
                    job_id = path_parts[3]
                    if job_id in SimpleForensicsHandler.jobs and SimpleForensicsHandler.jobs[job_id]["status"] == "completed":
                        mock_results = self.generate_mock_results()
                        response = {
                            "job_id": job_id,
                            "total_results": len(mock_results),
                            "results": mock_results
                        }
                        self.send_json_response(response)
                    else:
                        self.send_json_error(404, "Results not available")
                else:
                    self.send_json_error(400, "Invalid job ID")
            
            elif self.path == "/api/health":
                # Health check
                response = {
                    "status": "healthy",
                    "app_name": "VDF iOS Forensics",
                    "version": "1.0.0",
                    "active_jobs": len(SimpleForensicsHandler.jobs)
                }
                self.send_json_response(response)
                    
            else:
                self.send_json_error(404, f"API endpoint not found: {self.path}")
                
        except Exception as e:
            logging.error(f"API error: {e}")
            import traceback
            traceback.print_exc()
            self.send_json_error(500, str(e))
    
    def send_json_response(self, data, status_code=200):
        """Send JSON response."""
        try:
            response = json.dumps(data, default=str).encode('utf-8')
            self.send_response(status_code)
            self.send_header("Content-Type", "application/json")
            self.send_header("Content-Length", str(len(response)))
            self.send_header("Access-Control-Allow-Origin", "*")
            self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS")
            self.send_header("Access-Control-Allow-Headers", "Content-Type")
            self.end_headers()
            self.wfile.write(response)
        except Exception as e:
            print(f"Error sending JSON response: {e}")
            self.send_error(500, "Internal server error")
    
    def send_json_error(self, status_code, message):
        """Send JSON error response."""
        error_data = {"error": message, "status_code": status_code}
        self.send_json_response(error_data, status_code)
    
    def simulate_job_progress(self, job_id):
        """Simulate job progress for demonstration."""
        import time
        
        try:
            print(f"Starting job simulation for {job_id}")
            
            # Simulate progress stages
            stages = [
                (10, "Initializing analysis..."),
                (25, "Decrypting backup..."),
                (40, "Extracting profile events..."),
                (60, "Analyzing artifacts..."),
                (80, "Correlating with threat intelligence..."),
                (100, "Analysis complete")
            ]
            
            for progress, message in stages:
                time.sleep(3)  # Wait 3 seconds between updates for better demo
                if job_id in SimpleForensicsHandler.jobs:
                    SimpleForensicsHandler.jobs[job_id]["progress"] = float(progress)
                    if progress < 100:
                        SimpleForensicsHandler.jobs[job_id]["status"] = "running"
                    else:
                        SimpleForensicsHandler.jobs[job_id]["status"] = "completed"
                        SimpleForensicsHandler.jobs[job_id]["results_count"] = 15
                        SimpleForensicsHandler.jobs[job_id]["threats_detected"] = 3
                    
                    print(f"Job {job_id}: {progress}% - {message}")
                        
        except Exception as e:
            logging.error(f"Job simulation error: {e}")
            if job_id in SimpleForensicsHandler.jobs:
                SimpleForensicsHandler.jobs[job_id]["status"] = "failed"
                SimpleForensicsHandler.jobs[job_id]["error_message"] = str(e)
    
    def generate_mock_results(self):
        """Generate mock forensic results for demonstration."""
        return [
            {
                "artifact_type": "ios_profile_event",
                "timestamp": "2024-01-15T10:30:00Z",
                "data": {
                    "profile_id": "com.example.mdm.profile",
                    "operation": "install",
                    "process": "profiled"
                },
                "threat_detected": False
            },
            {
                "artifact_type": "ios_profile_event", 
                "timestamp": "2024-01-15T10:35:00Z",
                "data": {
                    "profile_id": "suspicious.malware.profile",
                    "operation": "install",
                    "process": "unknown_process"
                },
                "threat_detected": True,
                "threat_details": {
                    "indicator_name": "Suspicious Profile",
                    "confidence": 0.85
                }
            },
            {
                "artifact_type": "ios_profile_event",
                "timestamp": "2024-01-15T11:00:00Z", 
                "data": {
                    "profile_id": "com.enterprise.vpn.profile",
                    "operation": "remove",
                    "process": "Settings"
                },
                "threat_detected": False
            }
        ]
    
    def log_message(self, format, *args):
        """Override to customize logging."""
        # Log API requests for debugging
        if self.path.startswith("/api/"):
            print(f"API: {format % args}")
        elif self.path in ["/", "/index.html", "/app"]:
            print(f"UI: {format % args}")
        # Skip logging for static files to reduce noise


def create_static_files():
    """Ensure static files exist."""
    STATIC_DIR.mkdir(parents=True, exist_ok=True)
    
    # Create a simple index.html if it doesn't exist
    index_file = STATIC_DIR / "index.html"
    if not index_file.exists():
        # Copy the existing one or create a minimal version
        try:
            existing_index = Path(__file__).parent / "app" / "static" / "index.html"
            if existing_index.exists():
                import shutil
                shutil.copy(existing_index, index_file)
            else:
                # Create minimal HTML
                with open(index_file, 'w') as f:
                    f.write("""
<!DOCTYPE html>
<html>
<head><title>VDF iOS Forensics</title></head>
<body>
    <h1>VDF iOS Forensics Toolkit</h1>
    <p>Development server is running!</p>
    <p>The full interface should be available if static files are properly set up.</p>
</body>
</html>
""")
        except Exception as e:
            logging.warning(f"Could not create static files: {e}")


def main():
    """Main entry point."""
    logging.basicConfig(
        level=logging.INFO,
        format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
    )
    
    logger = logging.getLogger(__name__)
    
    # Create static files
    create_static_files()
    
    # Start server
    logger.info(f"Starting VDF iOS Forensics development server...")
    logger.info(f"Server will be available at: http://{SERVER_HOST}:{SERVER_PORT}")
    logger.info(f"Static files served from: {STATIC_DIR}")
    
    try:
        server = HTTPServer((SERVER_HOST, SERVER_PORT), SimpleForensicsHandler)
        
        # Open browser
        threading.Timer(1.0, lambda: webbrowser.open(f"http://{SERVER_HOST}:{SERVER_PORT}")).start()
        
        logger.info("Server started successfully. Press Ctrl+C to stop.")
        server.serve_forever()
        
    except KeyboardInterrupt:
        logger.info("Server stopped by user")
    except Exception as e:
        logger.error(f"Server error: {e}")
        sys.exit(1)


if __name__ == "__main__":
    main()