""" EU-Utility Integration Test - Browser Extension ================================================ Tests browser extension communication protocols: - Native messaging host - WebSocket bridge - HTTP API endpoints - Message serialization Author: Integration Tester Version: 1.0.0 """ import json import struct import sys import os from typing import Dict, Any, Optional, List from dataclasses import dataclass, asdict from pathlib import Path from plugins.base_plugin import BasePlugin from PyQt6.QtWidgets import ( QWidget, QVBoxLayout, QHBoxLayout, QLabel, QLineEdit, QTextEdit, QPushButton, QComboBox, QGroupBox, QTabWidget, QTableWidget, QTableWidgetItem, QHeaderView, QFileDialog, QMessageBox, QCheckBox ) from PyQt6.QtCore import Qt, QThread, pyqtSignal @dataclass class BrowserProtocol: """Browser communication protocol definition.""" name: str protocol: str # 'native_messaging', 'websocket', 'http' browser: str # 'chrome', 'firefox', 'edge' description: str class BrowserExtensionTester(BasePlugin): """Plugin for testing browser extension integration.""" name = "Browser Extension Tester" version = "1.0.0" author = "Integration Tester" description = "Test browser extension communication protocols" # Native messaging manifest templates CHROME_MANIFEST = { "name": "eu_utility_native_host", "description": "EU-Utility Native Messaging Host", "path": "", # To be filled "type": "stdio", "allowed_origins": [ "chrome-extension://*/" ] } FIREFOX_MANIFEST = { "name": "eu_utility_native_host", "description": "EU-Utility Native Messaging Host", "path": "", # To be filled "type": "stdio", "allowed_extensions": [ "eu-utility@impulsivefps" ] } # Test message types TEST_MESSAGES = [ { "name": "Ping", "message": {"type": "ping", "timestamp": 1234567890}, "description": "Basic connectivity check" }, { "name": "Get Status", "message": {"type": "get_status"}, "description": "Request current status" }, { "name": "Search Request", "message": { "type": "search", "query": "ArMatrix LP-35", "entity_type": "weapons" }, "description": "Search via Nexus API" }, { "name": "Loot Event", "message": { "type": "loot_event", "data": { "mob": "Atrox Young", "value": 50.25, "items": ["Animal Oil", "Animal Hide"] } }, "description": "Forward loot event" }, { "name": "Global Announcement", "message": { "type": "global", "data": { "value": 150.0, "mob": "Atrox Alpha", "player": "TestPlayer" } }, "description": "Global/HOF announcement" }, ] def initialize(self): """Initialize the tester.""" self.log_info("Browser Extension Tester initialized") self._native_host_path = "" self._websocket_port = 8765 self._http_port = 8080 self._message_log: List[Dict] = [] def get_ui(self) -> QWidget: """Create the plugin UI.""" widget = QWidget() layout = QVBoxLayout(widget) # Title title = QLabel("Browser Extension Integration Tester") title.setStyleSheet("font-size: 16px; font-weight: bold;") layout.addWidget(title) desc = QLabel("Test browser extension communication via Native Messaging, WebSocket, and HTTP") desc.setWordWrap(True) layout.addWidget(desc) # Tabs tabs = QTabWidget() # Setup tab tabs.addTab(self._create_setup_tab(), "Setup") # Native Messaging tab tabs.addTab(self._create_native_messaging_tab(), "Native Messaging") # WebSocket tab tabs.addTab(self._create_websocket_tab(), "WebSocket") # HTTP API tab tabs.addTab(self._create_http_tab(), "HTTP API") # Message Tester tab tabs.addTab(self._create_message_tab(), "Message Tester") # Log tab tabs.addTab(self._create_log_tab(), "Message Log") layout.addWidget(tabs) return widget def _create_setup_tab(self) -> QWidget: """Create the setup tab.""" widget = QWidget() layout = QVBoxLayout(widget) # Browser selection browser_group = QGroupBox("Browser Configuration") browser_layout = QVBoxLayout(browser_group) browser_row = QHBoxLayout() browser_row.addWidget(QLabel("Browser:")) self.browser_select = QComboBox() self.browser_select.addItems(["Chrome", "Firefox", "Edge"]) browser_row.addWidget(self.browser_select) browser_layout.addLayout(browser_row) # Native host path host_row = QHBoxLayout() host_row.addWidget(QLabel("Native Host Path:")) self.host_path_input = QLineEdit() self.host_path_input.setPlaceholderText("Path to native messaging host executable") host_row.addWidget(self.host_path_input) browse_btn = QPushButton("Browse...") browse_btn.clicked.connect(self._browse_host_path) host_row.addWidget(browse_btn) browser_layout.addLayout(host_row) # Generate manifest button manifest_btn = QPushButton("Generate Manifest") manifest_btn.clicked.connect(self._generate_manifest) browser_layout.addWidget(manifest_btn) layout.addWidget(browser_group) # Installation instructions instructions = QTextEdit() instructions.setReadOnly(True) instructions.setHtml("""

Browser Extension Setup

Chrome/Edge:

  1. Generate the native messaging host manifest
  2. Copy to: %LOCALAPPDATA%\\Google\\Chrome\\User Data\\NativeMessagingHosts/
  3. Install the browser extension

Firefox:

  1. Generate the native messaging host manifest
  2. Copy to: %APPDATA%\\Mozilla\\NativeMessagingHosts/
  3. Install the browser extension
""") layout.addWidget(instructions) layout.addStretch() return widget def _create_native_messaging_tab(self) -> QWidget: """Create the native messaging tab.""" widget = QWidget() layout = QVBoxLayout(widget) layout.addWidget(QLabel("Native Messaging Protocol Test")) # Status self.native_status = QLabel("Status: Not connected") layout.addWidget(self.native_status) # Test messages layout.addWidget(QLabel("Test Messages:")) self.native_msg_list = QComboBox() for msg in self.TEST_MESSAGES: self.native_msg_list.addItem(f"{msg['name']}: {msg['description']}") layout.addWidget(self.native_msg_list) # Message preview layout.addWidget(QLabel("Message Preview:")) self.native_preview = QTextEdit() self.native_preview.setReadOnly(True) self.native_preview.setMaximumHeight(100) layout.addWidget(self.native_preview) # Update preview on selection change self.native_msg_list.currentIndexChanged.connect(self._update_native_preview) self._update_native_preview(0) # Send button send_btn = QPushButton("Send Test Message") send_btn.clicked.connect(self._send_native_message) layout.addWidget(send_btn) # Simulate browser response sim_btn = QPushButton("Simulate Browser Response") sim_btn.clicked.connect(self._simulate_browser_response) layout.addWidget(sim_btn) # Protocol info info = QTextEdit() info.setReadOnly(True) info.setHtml("""

Native Messaging Protocol

Messages are sent over stdin/stdout with a 32-bit length prefix:

[length: 4 bytes][JSON message: N bytes]

All messages must be valid JSON.

""") layout.addWidget(info) return widget def _create_websocket_tab(self) -> QWidget: """Create the WebSocket tab.""" widget = QWidget() layout = QVBoxLayout(widget) layout.addWidget(QLabel("WebSocket Bridge Test")) # Port configuration port_row = QHBoxLayout() port_row.addWidget(QLabel("WebSocket Port:")) self.ws_port_input = QLineEdit("8765") port_row.addWidget(self.ws_port_input) layout.addLayout(port_row) # Control buttons btn_layout = QHBoxLayout() self.start_ws_btn = QPushButton("Start Server") self.start_ws_btn.clicked.connect(self._start_websocket) btn_layout.addWidget(self.start_ws_btn) self.stop_ws_btn = QPushButton("Stop Server") self.stop_ws_btn.clicked.connect(self._stop_websocket) self.stop_ws_btn.setEnabled(False) btn_layout.addWidget(self.stop_ws_btn) layout.addLayout(btn_layout) # Status self.ws_status = QLabel("Status: Server not running") layout.addWidget(self.ws_status) # Connected clients self.ws_clients = QLabel("Connected clients: 0") layout.addWidget(self.ws_clients) # Message tester layout.addWidget(QLabel("Send Message to All Clients:")) self.ws_message = QTextEdit() self.ws_message.setPlaceholderText('{"type": "test", "data": "Hello from EU-Utility"}') self.ws_message.setMaximumHeight(80) layout.addWidget(self.ws_message) broadcast_btn = QPushButton("Broadcast") broadcast_btn.clicked.connect(self._broadcast_websocket) layout.addWidget(broadcast_btn) # Received messages layout.addWidget(QLabel("Received Messages:")) self.ws_received = QTextEdit() self.ws_received.setReadOnly(True) layout.addWidget(self.ws_received) return widget def _create_http_tab(self) -> QWidget: """Create the HTTP API tab.""" widget = QWidget() layout = QVBoxLayout(widget) layout.addWidget(QLabel("HTTP API Bridge Test")) # Port configuration port_row = QHBoxLayout() port_row.addWidget(QLabel("HTTP Port:")) self.http_port_input = QLineEdit("8080") port_row.addWidget(self.http_port_input) layout.addLayout(port_row) # Control buttons btn_layout = QHBoxLayout() self.start_http_btn = QPushButton("Start Server") self.start_http_btn.clicked.connect(self._start_http) btn_layout.addWidget(self.start_http_btn) self.stop_http_btn = QPushButton("Stop Server") self.stop_http_btn.clicked.connect(self._stop_http) self.stop_http_btn.setEnabled(False) btn_layout.addWidget(self.stop_http_btn) layout.addLayout(btn_layout) # Status self.http_status = QLabel("Status: Server not running") layout.addWidget(self.http_status) # Endpoints list layout.addWidget(QLabel("Available Endpoints:")) endpoints = QTextEdit() endpoints.setReadOnly(True) endpoints.setHtml("""
MethodEndpointDescription
GET/healthHealth check
GET/api/statusGet EU-Utility status
POST/api/notifySend notification
POST/api/searchSearch Nexus
GET/api/loot/sessionGet loot session
""") layout.addWidget(endpoints) # Test request layout.addWidget(QLabel("Test Request:")) test_layout = QHBoxLayout() self.http_method = QComboBox() self.http_method.addItems(["GET", "POST"]) test_layout.addWidget(self.http_method) self.http_endpoint = QLineEdit("/api/status") test_layout.addWidget(self.http_endpoint) test_btn = QPushButton("Send") test_btn.clicked.connect(self._test_http_request) test_layout.addWidget(test_btn) layout.addLayout(test_layout) # Response layout.addWidget(QLabel("Response:")) self.http_response = QTextEdit() self.http_response.setReadOnly(True) layout.addWidget(self.http_response) return widget def _create_message_tab(self) -> QWidget: """Create the message tester tab.""" widget = QWidget() layout = QVBoxLayout(widget) layout.addWidget(QLabel("Message Protocol Tester")) # Message type msg_row = QHBoxLayout() msg_row.addWidget(QLabel("Message Type:")) self.msg_type = QComboBox() self.msg_type.addItems(["Native", "WebSocket", "HTTP"]) msg_row.addWidget(self.msg_type) layout.addLayout(msg_row) # JSON input layout.addWidget(QLabel("JSON Message:")) self.msg_input = QTextEdit() self.msg_input.setPlaceholderText('Enter JSON message here...') layout.addWidget(self.msg_input) # Validate button validate_btn = QPushButton("Validate JSON") validate_btn.clicked.connect(self._validate_json) layout.addWidget(validate_btn) # Encode/Decode btn_layout = QHBoxLayout() encode_btn = QPushButton("Encode for Native Messaging") encode_btn.clicked.connect(self._encode_native) btn_layout.addWidget(encode_btn) decode_btn = QPushButton("Decode Native Message") decode_btn.clicked.connect(self._decode_native) btn_layout.addWidget(decode_btn) layout.addLayout(btn_layout) # Result layout.addWidget(QLabel("Result:")) self.msg_result = QTextEdit() self.msg_result.setReadOnly(True) layout.addWidget(self.msg_result) # Preset buttons layout.addWidget(QLabel("Load Preset:")) preset_layout = QHBoxLayout() for msg in self.TEST_MESSAGES: btn = QPushButton(msg['name']) btn.clicked.connect(lambda checked, m=msg: self._load_preset(m)) preset_layout.addWidget(btn) layout.addLayout(preset_layout) return widget def _create_log_tab(self) -> QWidget: """Create the message log tab.""" widget = QWidget() layout = QVBoxLayout(widget) layout.addWidget(QLabel("Message Log")) # Log table self.log_table = QTableWidget() self.log_table.setColumnCount(4) self.log_table.setHorizontalHeaderLabels(["Time", "Direction", "Protocol", "Message"]) self.log_table.horizontalHeader().setSectionResizeMode(3, QHeaderView.ResizeMode.Stretch) layout.addWidget(self.log_table) # Buttons btn_layout = QHBoxLayout() clear_btn = QPushButton("Clear Log") clear_btn.clicked.connect(self._clear_log) btn_layout.addWidget(clear_btn) export_btn = QPushButton("Export Log") export_btn.clicked.connect(self._export_log) btn_layout.addWidget(export_btn) layout.addLayout(btn_layout) return widget def _browse_host_path(self): """Browse for native host executable.""" path, _ = QFileDialog.getOpenFileName( self.get_ui(), "Select Native Host Executable", "", "Executables (*.exe *.py *.sh);;All Files (*)" ) if path: self.host_path_input.setText(path) self._native_host_path = path def _generate_manifest(self): """Generate native messaging manifest.""" browser = self.browser_select.currentText() host_path = self.host_path_input.text() if not host_path: QMessageBox.warning(self.get_ui(), "Missing Path", "Please select the native host executable path") return # Select appropriate template if browser in ["Chrome", "Edge"]: manifest = self.CHROME_MANIFEST.copy() else: manifest = self.FIREFOX_MANIFEST.copy() manifest["path"] = host_path # Get save location save_path, _ = QFileDialog.getSaveFileName( self.get_ui(), "Save Manifest", f"eu_utility_native_host.json", "JSON Files (*.json)" ) if save_path: with open(save_path, 'w') as f: json.dump(manifest, f, indent=2) self.notify_success("Manifest Generated", f"Saved to: {save_path}") def _update_native_preview(self, index: int): """Update native messaging preview.""" if 0 <= index < len(self.TEST_MESSAGES): msg = self.TEST_MESSAGES[index] self.native_preview.setText(json.dumps(msg['message'], indent=2)) def _send_native_message(self): """Send a test native messaging message.""" index = self.native_msg_list.currentIndex() if index < 0: return msg = self.TEST_MESSAGES[index] self._log_message("OUT", "native", msg['message']) # Simulate sending self.notify_info("Message Sent", f"Sent: {msg['name']}") def _simulate_browser_response(self): """Simulate a browser response.""" response = { "type": "response", "id": "test-123", "success": True, "data": {"message": "Browser received the message"} } self._log_message("IN", "native", response) self.notify_info("Response Simulated", "Browser response added to log") def _start_websocket(self): """Start WebSocket server.""" self.notify_info("WebSocket", "Starting WebSocket server...") self.ws_status.setText("Status: Starting...") self.start_ws_btn.setEnabled(False) self.stop_ws_btn.setEnabled(True) def _stop_websocket(self): """Stop WebSocket server.""" self.notify_info("WebSocket", "Stopping WebSocket server...") self.ws_status.setText("Status: Server not running") self.start_ws_btn.setEnabled(True) self.stop_ws_btn.setEnabled(False) def _broadcast_websocket(self): """Broadcast message to WebSocket clients.""" message = self.ws_message.toPlainText() if not message: return try: msg_data = json.loads(message) self._log_message("OUT", "websocket", msg_data) self.notify_info("Broadcast", "Message broadcast to clients") except json.JSONDecodeError: self.notify_error("Invalid JSON", "Message is not valid JSON") def _start_http(self): """Start HTTP server.""" self.notify_info("HTTP", "Starting HTTP API server...") self.http_status.setText("Status: Running on http://localhost:8080") self.start_http_btn.setEnabled(False) self.stop_http_btn.setEnabled(True) def _stop_http(self): """Stop HTTP server.""" self.notify_info("HTTP", "Stopping HTTP API server...") self.http_status.setText("Status: Server not running") self.start_http_btn.setEnabled(True) self.stop_http_btn.setEnabled(False) def _test_http_request(self): """Test HTTP request.""" method = self.http_method.currentText() endpoint = self.http_endpoint.text() self.http_response.setText(f"Simulated {method} {endpoint}\n\nResponse: 200 OK") def _validate_json(self): """Validate JSON input.""" text = self.msg_input.toPlainText() try: data = json.loads(text) self.msg_result.setText(f"✅ Valid JSON\n\n{json.dumps(data, indent=2)}") except json.JSONDecodeError as e: self.msg_result.setText(f"❌ Invalid JSON: {str(e)}") def _encode_native(self): """Encode message for native messaging.""" text = self.msg_input.toPlainText() try: data = json.loads(text) message = json.dumps(data) # Native messaging format: 32-bit length prefix + JSON encoded = struct.pack('I', len(message)) + message.encode('utf-8') self.msg_result.setText(f"Encoded ({len(encoded)} bytes):\n{encoded.hex()}") except json.JSONDecodeError: self.msg_result.setText("❌ Invalid JSON") def _decode_native(self): """Decode native messaging format.""" # This would decode hex input back to JSON self.msg_result.setText("Decode functionality - paste hex bytes to decode") def _load_preset(self, msg: Dict): """Load a message preset.""" self.msg_input.setText(json.dumps(msg['message'], indent=2)) def _log_message(self, direction: str, protocol: str, message: Dict): """Log a message.""" from datetime import datetime entry = { "time": datetime.now().strftime("%H:%M:%S"), "direction": direction, "protocol": protocol, "message": json.dumps(message)[:100] } self._message_log.append(entry) # Update table row = self.log_table.rowCount() self.log_table.insertRow(row) self.log_table.setItem(row, 0, QTableWidgetItem(entry["time"])) self.log_table.setItem(row, 1, QTableWidgetItem(entry["direction"])) self.log_table.setItem(row, 2, QTableWidgetItem(entry["protocol"])) self.log_table.setItem(row, 3, QTableWidgetItem(entry["message"])) def _clear_log(self): """Clear message log.""" self._message_log.clear() self.log_table.setRowCount(0) def _export_log(self): """Export message log.""" if not self._message_log: self.notify_warning("Empty Log", "No messages to export") return filename = f"browser_protocol_log_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json" with open(filename, 'w') as f: json.dump(self._message_log, f, indent=2) self.notify_success("Exported", f"Log saved to {filename}") plugin_class = BrowserExtensionTester