convert_xml.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

import json
import sys
import base64
import xml.etree.ElementTree as ET
from datetime import datetime
from typing import Dict, List, Optional
from pathlib import Path


class BurpXMLToHAR:
    """Burp Suite XML to HAR Converter -100% Accuracy"""
    
    def __init__(self):
        self.entries = []
        self.stats = {
            'total_items': 0,
            'entries_created': 0,
            'errors': []
        }
    
    def parse_headers(self, header_text: str) -> List[Dict[str, str]]:
        """Analyze headers from HTTP"""
        headers = []
        lines = header_text.split('\r\n')  # 🔥 HTTP standard
        if not lines:
            lines = header_text.split('\n')  # fallback
        
        for line in lines[1:]:  # Skip the first line (request/response line)
            line = line.strip()
            if not line:
                break
            
            if ':' in line:
                name, value = line.split(':', 1)
                headers.append({
                    "name": name.strip(),
                    "value": value.strip()
                })
        
        return headers
    
    def parse_cookies(self, headers: List[Dict[str, str]]) -> List[Dict[str, str]]:
        """Extract cookies from headers"""
        cookies = []
        
        for h in headers:
            if h['name'].lower() == 'cookie':
                for cookie in h['value'].split(';'):
                    cookie = cookie.strip()
                    if '=' in cookie:
                        name, value = cookie.split('=', 1)
                        cookies.append({
                            "name": name.strip(),
                            "value": value.strip()
                        })
        
        return cookies
    
    def extract_set_cookies(self, headers: List[Dict[str, str]]) -> List[Dict[str, str]]:
        """Extract Set-Cookie from response"""
        cookies = []
        
        for h in headers:
            if h['name'].lower() == 'set-cookie':
                parts = h['value'].split(';')
                if parts and '=' in parts[0]:
                    name, value = parts[0].split('=', 1)
                    cookies.append({
                        "name": name.strip(),
                        "value": value.strip()
                    })
        
        return cookies
    
    def parse_query_string(self, url: str) -> List[Dict[str, str]]:
        """Extract query parameters"""
        params = []
        
        if '?' not in url:
            return params
        
        query = url.split('?', 1)[1]
        if '#' in query:
            query = query.split('#')[0]
        
        for param in query.split('&'):
            if '=' in param:
                name, value = param.split('=', 1)
                params.append({
                    "name": name,
                    "value": value
                })
            elif param:
                params.append({
                    "name": param,
                    "value": ""
                })
        
        return params
    
    def decode_base64_safe(self, data: str) -> str:
        """Decode base64 safely"""
        try:
            decoded = base64.b64decode(data).decode('utf-8', errors='replace')
            return decoded
        except Exception as e:
            self.stats['errors'].append(f"Base64 decode error: {e}")
            return ""
    
    def parse_http_request(self, request_text: str) -> tuple:
        """تحليل HTTP request"""
        lines = request_text.split('\r\n')  # 🔥 HTTP Used \r\n
        if not lines:
            return None, None, None, None, None
        
        # First line: GET /path HTTP/1.1
        first_line = lines[0].strip()
        parts = first_line.split(' ')
        
        if len(parts) < 3:
            return None, None, None, None, None
        
        method = parts[0]
        path = parts[1]
        http_version = parts[2]
        
        # Headers
        headers = self.parse_headers(request_text)
        
        # Body (everything after a blank line)
        body = None
        body_start = False
        body_lines = []
        
        for line in lines:
            if body_start:
                body_lines.append(line)
            elif not line.strip():
                body_start = True
        
        if body_lines:
            body = '\r\n'.join(body_lines).strip()
        
        return method, path, http_version, headers, body
    
    def parse_http_response(self, response_text: str) -> tuple:
        """تحليل HTTP response"""
        lines = response_text.split('\r\n')  # 🔥 HTTP Used \r\n
        if not lines:
            return None, None, None, None, None
        
        # First line: HTTP/1.1 200 OK
        first_line = lines[0].strip()
        parts = first_line.split(' ', 2)
        
        if len(parts) < 3:
            return None, None, None, None, None
        
        http_version = parts[0]
        status_code = int(parts[1])
        status_text = parts[2]
        
        # Headers
        headers = self.parse_headers(response_text)
        
        # Body
        body = None
        body_start = False
        body_lines = []
        
        for line in lines:
            if body_start:
                body_lines.append(line)
            elif not line.strip():
                body_start = True
        
        if body_lines:
            body = '\r\n'.join(body_lines).strip()
        
        return http_version, status_code, status_text, headers, body
    
    def parse_timestamp(self, time_str: str) -> str:
        """Convert timestamp from XML"""
        try:
            # Format: "Sat Jan 17 17:08:18 EET 2026"
            dt = datetime.strptime(time_str.strip(), "%a %b %d %H:%M:%S %Z %Y")
            return dt.isoformat() + 'Z'
        except:
            return datetime.now().isoformat() + 'Z'
    
    def parse_xml_file(self, filepath: str):
        """Read and analyze XML from Burp Suite"""
        
        try:
            tree = ET.parse(filepath)
            root = tree.getroot()
            
            items = root.findall('item')
            self.stats['total_items'] = len(items)
            
            print(f"📊 was found{len(items)} item In the file\n")
            
            for idx, item in enumerate(items, 1):
                try:
                    # Basic data extraction
                    time_elem = item.find('time')
                    url_elem = item.find('url')
                    host_elem = item.find('host')
                    port_elem = item.find('port')
                    protocol_elem = item.find('protocol')
                    method_elem = item.find('method')
                    request_elem = item.find('request')
                    status_elem = item.find('status')
                    response_elem = item.find('response')
                    mimetype_elem = item.find('mimetype')
                    
                    if url_elem is None or request_elem is None:
                        continue
                    
                    # Timestamp
                    timestamp = self.parse_timestamp(time_elem.text) if time_elem is not None else datetime.now().isoformat() + 'Z'
                    
                    # URL
                    url = url_elem.text.strip() if url_elem.text else ""
                    if not url:
                        continue
                    
                    # Host & IP
                    host = host_elem.text.strip() if host_elem is not None and host_elem.text else ""
                    ip = host_elem.get('ip', '') if host_elem is not None else ""
                    
                    # Port & Protocol
                    port = port_elem.text.strip() if port_elem is not None and port_elem.text else "443"
                    protocol = protocol_elem.text.strip() if protocol_elem is not None and protocol_elem.text else "https"
                    
                    # ===== Request Parsing =====
                    request_data = request_elem.text or ""
                    is_base64_request = request_elem.get('base64', 'false') == 'true'
                    
                    if is_base64_request:
                        request_data = self.decode_base64_safe(request_data)
                    
                    method, path, http_version, req_headers, req_body = self.parse_http_request(request_data)
                    
                    if not method:
                        method = method_elem.text.strip() if method_elem is not None and method_elem.text else "GET"
                    
                    # Query String
                    query_string = self.parse_query_string(url)
                    
                    # Cookies
                    cookies = self.parse_cookies(req_headers)
                    
                    # Content Type & Length
                    content_type = "application/octet-stream"
                    content_length = 0
                    
                    for h in req_headers:
                        if h['name'].lower() == 'content-type':
                            content_type = h['value'].split(';')[0].strip()
                        elif h['name'].lower() == 'content-length':
                            try:
                                content_length = int(h['value'])
                            except:
                                pass
                    
                    # 🔥 Actual body size calculation
                    actual_body_size = len(req_body.encode('utf-8')) if req_body else 0
                    
                    # Request Object
                    request_obj = {
                        "method": method,
                        "url": url,
                        "httpVersion": http_version or "HTTP/1.1",
                        "headers": req_headers,
                        "queryString": query_string,
                        "cookies": cookies,
                        "headersSize": -1,
                        "bodySize": actual_body_size  # 🔥 Actual size
                    }
                    
                    if req_body:
                        request_obj["postData"] = {
                            "mimeType": content_type,
                            "text": req_body
                        }
                    
                    # ===== Response Parsing =====
                    response_data = response_elem.text or "" if response_elem is not None else ""
                    is_base64_response = response_elem.get('base64', 'false') == 'true' if response_elem is not None else False
                    
                    if is_base64_response:
                        response_data = self.decode_base64_safe(response_data)
                    
                    resp_version, status_code, status_text, resp_headers, resp_body = self.parse_http_response(response_data)
                    
                    if status_code is None:
                        status_code = int(status_elem.text.strip()) if status_elem is not None and status_elem.text else 0
                        status_text = ""
                    
                    # Response Content Type
                    resp_content_type = mimetype_elem.text.strip() if mimetype_elem is not None and mimetype_elem.text else "text/html"
                    
                    for h in resp_headers:
                        if h['name'].lower() == 'content-type':
                            resp_content_type = h['value'].split(';')[0].strip()
                            break
                    
                    # Response Cookies
                    response_cookies = self.extract_set_cookies(resp_headers)
                    
                    # 🔥 Calculate the actual response size
                    actual_resp_size = len(resp_body.encode('utf-8')) if resp_body else 0
                    
                    # Response Object
                    response_obj = {
                        "status": status_code,
                        "statusText": status_text or "",
                        "httpVersion": resp_version or "HTTP/1.1",
                        "headers": resp_headers,
                        "cookies": response_cookies,
                        "content": {
                            "size": actual_resp_size,  # 🔥 Actual size
                            "mimeType": resp_content_type,
                            "text": resp_body or ""
                        },
                        "redirectURL": "",
                        "headersSize": -1,
                        "bodySize": actual_resp_size  # 🔥 Actual size
                    }
                    
                    # HAR Entry
                    entry = {
                        "startedDateTime": timestamp,
                        "time": 100,
                        "request": request_obj,
                        "response": response_obj,
                        "cache": {},
                        "timings": {
                            "blocked": 0,
                            "dns": 0,
                            "connect": 0,
                            "send": 1,
                            "wait": 50,
                            "receive": 49,
                            "ssl": 0
                        }
                    }
                    
                    if ip:
                        entry["serverIPAddress"] = ip
                    
                    self.entries.append(entry)
                    self.stats['entries_created'] += 1
                    
                    if idx % 10 == 0:
                        print(f"  ⚡ Process: {idx}/{len(items)} items...")
                
                except Exception as e:
                    self.stats['errors'].append(f"Item {idx} error: {e}")
                    continue
        
        except Exception as e:
            raise Exception(f"Failed to read XML: {e}")
    
    def generate_har(self) -> Dict:
        """HAR generation is 100% compliant with standards"""
        return {
            "log": {
                "version": "1.2",
                "creator": {
                    "name": "Burp XML to HAR Converter",
                    "version": "1.0",
                    "comment": "Converted from Burp Suite XML export"
                },
                "browser": {
                    "name": "Unknown",
                    "version": "Unknown"
                },
                "pages": [],
                "entries": self.entries,
                "comment": f"Converted {len(self.entries)} HTTP transactions from Burp Suite XML"
            }
        }
    
    def save_har(self, output_path: str):
        """Save HAR"""
        try:
            har = self.generate_har()
            
            with open(output_path, 'w', encoding='utf-8') as f:
                json.dump(har, f, indent=2, ensure_ascii=False)
            
            print(f"\n{'='*60}")
            print(f"✅Successfully saved!")
            print(f"{'='*60}")
            print(f"📁 File: {output_path}")
            print(f"📊 statistics:")
            print(f"   • total Items: {self.stats['total_items']}")
            print(f"   • Entries Facility: {self.stats['entries_created']}")
            
            if self.stats['errors']:
                print(f"\n⚠️  warnings ({len(self.stats['errors'])}):")
                for err in self.stats['errors'][:5]:
                    print(f"   • {err}")
                if len(self.stats['errors']) > 5:
                    print(f"   ... and {len(self.stats['errors']) - 5}Other warnings")
        
        except Exception as e:
            raise Exception(f"Failed to save HAR: {e}")


def main():
    print("=" * 60)
    print("  📄 Burp Suite XML to HAR Converter v1.0")
    print(" Burp XML to HAR Converter -100% Accuracy")
    print("=" * 60)
    
    if len(sys.argv) < 2:
        print("\n📖 Usage:")
        print("  python burp_to_har.py <input.xml> [output.har]\n")
        print("💡 Examples:")
        print("  python burp_to_har.py burp_export.xml")
        print("  python burp_to_har.py burp_export.xml output.har\n")
        sys.exit(1)
    
    input_file = sys.argv[1]
    output_file = sys.argv[2] if len(sys.argv) > 2 else "output.har"
    
    if not Path(input_file).exists():
        print(f"\n❌File not found: {input_file}")
        sys.exit(1)
    
    file_size = Path(input_file).stat().st_size
    print(f"\n📂 Input file: {input_file}")
    print(f"📏 File size: {file_size:,} bytes ({file_size/1024/1024:.2f} MB)")
    print(f"💾 Output file: {output_file}\n")
    
    converter = BurpXMLToHAR()
    
    try:
        print("🔍 Start analysis...\n")
        converter.parse_xml_file(input_file)
        
        if len(converter.entries) == 0:
            print("\n⚠️ Warning: No valid entries found!")
            sys.exit(1)
        
        converter.save_har(output_file)
        
        print(f"\n{'='*60}")
        print("🎯 Next steps:")
        print("   1. Open OWASP ZAP")
        print("   2. Go to: File → Import HAR file")
        print(f"   3. Chose: {output_file}")
        print("   4. All Requests will appear in History tab")
        print(f"{'='*60}\n")
    
    except Exception as e:
        print(f"\n❌ Error: {e}")
        import traceback
        print("\n📋 Error details:")
        traceback.print_exc()
        sys.exit(1)


if __name__ == "__main__":
    main()