""" Utility functions for backend plugins """ import logging import re from pathlib import Path from typing import Any from fastapi_app.config import get_settings logger = logging.getLogger(__name__) def get_plugin_config( config_key: str, env_var: str, default: Any = None, value_type: str = "string" ) -> Any: """ Get plugin configuration value with env var fallback. Priority: config.json > environment variable > default Creates config key from env var if it doesn't exist. Args: config_key: Dot-notation config key (e.g., "plugin.local-sync.enabled") env_var: Environment variable name default: Default value if neither source has value value_type: Type for validation ("string", "boolean", "number", "array") Returns: Configuration value Example: >>> enabled = get_plugin_config( ... "plugin.local-sync.enabled", ... "PLUGIN_LOCAL_SYNC_ENABLED", ... default=False, ... value_type="boolean" ... ) """ from fastapi_app.lib.utils.config_utils import get_config import os config = get_config() # Try to get from config value = config.get(config_key) if value is None: # Check environment variable env_value = os.environ.get(env_var) if env_value is not None: # Parse env value based on type if value_type == "boolean": value = env_value.lower() in ("true", "1", "yes") elif value_type == "number": value = int(env_value) if env_value.isdigit() else float(env_value) elif value_type == "array": import json value = json.loads(env_value) else: value = env_value # Create config key from env var config.set(config_key, value) else: # Use default value = default if value is not None: config.set(config_key, value) return value def _extract_sandbox_methods() -> list[dict[str, str]]: """ Extract public method signatures from PluginSandbox class. Reads backend-plugin-sandbox.js and parses JSDoc comments and method signatures to build a list of available public methods. Returns: List of dicts with 'name', 'params', and 'doc' keys Example return: [ { 'name': 'openDocument', 'params': ['stableId'], 'doc': 'Open a document by updating xml state and closing dialog' }, ... ] """ # Find the sandbox module file sandbox_file = get_settings().project_root_dir / 'app' / 'src' / 'modules' / 'backend-plugin-sandbox.js' if not sandbox_file.exists(): # Fallback to empty list if file not found return [] content = sandbox_file.read_text(encoding='utf-8') methods = [] # Pattern to match JSDoc comment followed by method definition # Matches: /** ... */ async methodName(param1, param2) { pattern = r'/\*\*\s*(.*?)\s*\*/\s*(?:async\s+)?(\w+)\s*\((.*?)\)\s*\{' for match in re.finditer(pattern, content, re.DOTALL): jsdoc = match.group(1) method_name = match.group(2) params_str = match.group(3) # Skip constructor and private methods if method_name == 'constructor' or method_name.startswith('_'): continue # Extract parameter names (ignore types and defaults) params = [] if params_str.strip(): for param in params_str.split(','): param = param.strip() # Extract just the name (before = or :) param_name = re.split(r'[=:]', param)[0].strip() if param_name: params.append(param_name) # Extract first line of JSDoc as description doc = '' for line in jsdoc.split('\n'): line = line.strip().lstrip('*').strip() # Skip @param, @returns, etc. if line and not line.startswith('@'): doc = line break methods.append({ 'name': method_name, 'params': params, 'doc': doc }) return methods def generate_sandbox_client_script() -> str: """ Generate JavaScript code that establishes connection with parent window and provides sandbox API access. Dynamically generates client methods based on PluginSandbox class definition. Returns: JavaScript code as string to be embedded in ''' """ # Extract available methods methods = _extract_sandbox_methods() if not methods: logger.warning("No sandbox methods extracted from PluginSandbox class. " "Sandbox client will have no methods available.") # Generate method wrappers method_code = [] for method in methods: params_str = ', '.join(method['params']) method_code.append(f""" /** * {method['doc']} * @param {{{', '.join(f'{p}' for p in method['params'])}}} */ {method['name']}({params_str}) {{ return callSandboxMethod('{method['name']}', {params_str}); }}""") methods_js = ',\n'.join(method_code) return f""" // Sandbox client for inter-window communication // Auto-generated from PluginSandbox class definition (function() {{ 'use strict'; // Determine parent window (iframe uses parent, popup uses opener) const parentWindow = window.parent !== window ? window.parent : window.opener; if (!parentWindow) {{ console.warn('SandboxClient: No parent window found. Page must be opened from within the application.'); // Create stub sandbox so callers get a descriptive error instead of "undefined" const notAvailable = (method) => () => {{ throw new Error('Sandbox not available: this page must be opened from within the application (no parent window).'); }}; window.sandbox = new Proxy({{}}, {{ get: (_, prop) => notAvailable(prop) }}); return; }} let requestId = 0; const pendingRequests = new Map(); // Listen for responses from parent window.addEventListener('message', (event) => {{ if (!event.data || event.data.type !== 'SANDBOX_RESPONSE') {{ return; }} const {{ requestId: respId, result, error }} = event.data; const pending = pendingRequests.get(respId); if (!pending) return; pendingRequests.delete(respId); if (error) {{ pending.reject(new Error(error)); }} else {{ pending.resolve(result); }} }}); /** * Call sandbox method in parent window * @param {{string}} method - Sandbox method name * @param {{...any}} args - Method arguments * @returns {{Promise}} Method result */ function callSandboxMethod(method, ...args) {{ return new Promise((resolve, reject) => {{ const reqId = requestId++; pendingRequests.set(reqId, {{ resolve, reject }}); // Send command to parent (iframe or opener) parentWindow.postMessage({{ type: 'SANDBOX_COMMAND', method, args, requestId: reqId }}, '*'); // Timeout after 10 seconds setTimeout(() => {{ if (pendingRequests.has(reqId)) {{ pendingRequests.delete(reqId); reject(new Error(`Request timeout: ${{method}}`)); }} }}, 10000); }}); }} // Expose sandbox API with dynamically generated methods window.sandbox = {{{methods_js} }}; // SSE event forwarding support (function() {{ const sseCallbacks = {{}}; // Listen for forwarded SSE events from parent window.addEventListener('message', (event) => {{ if (!event.data || event.data.type !== 'SSE_EVENT') return; const {{ subscriptionId, data }} = event.data; const cbs = sseCallbacks[subscriptionId]; if (cbs) {{ cbs.forEach(cb => {{ try {{ cb(JSON.parse(data)); }} catch {{ cb(data); }} }}); }} }}); // Override subscribeSSE to accept a callback parameter const origSubscribe = window.sandbox.subscribeSSE; if (origSubscribe) {{ window.sandbox.subscribeSSE = async function(eventType, callback) {{ const subId = await origSubscribe.call(window.sandbox, eventType); if (!sseCallbacks[subId]) sseCallbacks[subId] = []; sseCallbacks[subId].push(callback); return subId; }}; }} else {{ console.error('SandboxClient: subscribeSSE method not available. Sandbox methods may not have been generated correctly.'); }} // Override unsubscribeSSE to clean up local callbacks const origUnsubscribe = window.sandbox.unsubscribeSSE; if (origUnsubscribe) {{ window.sandbox.unsubscribeSSE = async function(subscriptionId) {{ delete sseCallbacks[subscriptionId]; return origUnsubscribe.call(window.sandbox, subscriptionId); }}; }} }})(); console.log('SandboxClient: Connected to parent window'); }})(); """.strip() def wrap_html_with_sandbox_client(html_content: str) -> str: """ Wrap HTML content with sandbox client script. Args: html_content: HTML content (can be partial or complete document) Returns: Complete HTML document with sandbox client script """ script = generate_sandbox_client_script() # If already complete document, inject script into head if ' or after if '' in html_content: return html_content.replace('', f'\n', 1) elif '' in html_content: return html_content.replace('', f'\n', 1) else: # No head tag, add it return html_content.replace('\n', 1) else: # Partial HTML, wrap it return f""" {html_content} """ def load_plugin_html( caller_file: str, template_name: str, inject_sandbox: bool = True, ) -> str: """ Load an HTML template from a plugin's ``static/`` directory (preferred) or the legacy ``html/`` directory. Looks for the template in ``static/`` first; falls back to ``html/`` for backwards compatibility (deprecated, prefer ``static/``). If *inject_sandbox* is True, the sandbox client script is injected into the ```` via :func:`wrap_html_with_sandbox_client`. Args: caller_file: ``__file__`` of the calling module – used to locate the plugin's ``static/`` (or ``html/``) directory. template_name: File name inside ``static/`` (e.g. ``"view.html"``). inject_sandbox: Inject the sandbox client script (default True). Returns: HTML string ready to be returned as ``HTMLResponse``. Example:: from fastapi_app.lib.plugins.plugin_tools import load_plugin_html @router.get("/view", response_class=HTMLResponse) async def view(...): html = load_plugin_html(__file__, "view.html") return HTMLResponse(content=html) """ import warnings plugin_dir = Path(caller_file).resolve().parent # Preferred: static/ template_path = plugin_dir / "static" / template_name if not template_path.exists(): # Fallback: html/ (deprecated) legacy_path = plugin_dir / "html" / template_name if legacy_path.exists(): warnings.warn( f"Plugin HTML template found in deprecated 'html/' directory: {legacy_path}. " "Move it to 'static/' instead.", DeprecationWarning, stacklevel=2, ) template_path = legacy_path else: raise FileNotFoundError( f"Plugin HTML template not found in 'static/' or 'html/': {template_name}" ) html = template_path.read_text(encoding="utf-8") if inject_sandbox: html = wrap_html_with_sandbox_client(html) return html def escape_html(text: str) -> str: """ Escape HTML special characters. Args: text: Text to escape Returns: Escaped text safe for HTML insertion """ if not text: return "" return ( text.replace("&", "&") .replace("<", "<") .replace(">", ">") .replace('"', """) .replace("'", "'") ) def validate_javascript_content(content: str, filename: str) -> tuple[bool, list[str]]: """ Validate JavaScript content for dangerous patterns. Checks for patterns that could introduce security risks: - Network access (fetch, XMLHttpRequest, WebSocket) - Dynamic code execution (eval, Function constructor) - Dynamic imports - Storage access (cookies, localStorage, sessionStorage, indexedDB) Args: content: JavaScript source code filename: Name of the file (for error messages) Returns: Tuple of (is_valid, list of warning messages) Example: >>> valid, warnings = validate_javascript_content(code, "my-extension.js") >>> if not valid: ... for warning in warnings: ... logger.warning(warning) """ warnings = [] # Strip comments before checking so JSDoc/TypeScript annotations # (e.g. @param {import('...')}) don't trigger false positives. # Order matters: block comments first, then line comments. code = re.sub(r'/\*.*?\*/', '', content, flags=re.DOTALL) code = re.sub(r'^\s*//[^\n]*', '', code, flags=re.MULTILINE) # Patterns that indicate potential security issues dangerous_patterns = [ (r'\bfetch\s*\(\s*(?![\'"`]/api/)', 'Network access via fetch() to external URLs'), (r'\bXMLHttpRequest\b', 'Network access via XMLHttpRequest'), (r'\bWebSocket\b', 'WebSocket connection'), (r'\beval\s*\(', 'Dynamic code execution via eval()'), (r'\bnew\s+Function\s*\(', 'Dynamic code execution via Function constructor'), (r'(? str: """ Generate a complete HTML page with a DataTables-powered table. Args: title: Page title headers: List of column headers rows: List of rows, each row is a list of cell values (can include HTML) table_id: HTML ID for the table element page_length: Number of rows per page default_sort_col: Column index to sort by default default_sort_dir: Sort direction ("asc" or "desc") enable_sandbox_client: Include sandbox client script for inter-window communication custom_css: Additional CSS to include in

{escape_html(title)}

{extra_content_before_table} {''.join(f'' for h in headers)} {''.join(rows_html)}
{escape_html(h)}
""" return html