From 88dc35889456f9b9e4d2c6f054b33ef58a835c95 Mon Sep 17 00:00:00 2001 From: claudi Date: Tue, 17 Feb 2026 15:56:53 +0100 Subject: [PATCH] Refactor drag handling and update tests - Renamed `initiate_drag` to `handle_drag` in MainWindow and updated related tests. - Improved drag handling logic to utilize a bridge for starting file drags. - Updated `_on_drag_started` and `_on_drag_failed` methods to match new signatures. - Modified test cases to reflect changes in drag handling and assertions. Enhance path validation and logging - Updated `PathValidator` to log warnings for nonexistent roots instead of raising errors. - Adjusted tests to verify the new behavior of skipping nonexistent roots. Update web application UI and functionality - Changed displayed text for drag items to reflect local paths and Azure Blob Storage URLs. - Added debug logging for drag operations in the web application. - Improved instructions for testing drag and drop functionality. Add configuration documentation and example files - Created `CONFIG_README.md` to provide detailed configuration instructions for WebDrop Bridge. - Added `config.example.json` and `config_test.json` for reference and testing purposes. Implement URL conversion logic - Introduced `URLConverter` class to handle conversion of Azure Blob Storage URLs to local paths. - Added unit tests for URL conversion to ensure correct functionality. Develop download interceptor script - Created `download_interceptor.js` to intercept download-related actions in the web application. - Implemented logging for fetch calls, XMLHttpRequests, and Blob URL creations. Add download test page and related tests - Created `test_download.html` for testing various download scenarios. - Implemented `test_download.py` to verify download path resolution and file construction. - Added `test_url_mappings.py` to ensure URL mappings are loaded correctly. Add unit tests for URL converter - Created `test_url_converter.py` to validate URL conversion logic and mapping behavior. --- CONFIG_README.md | 209 ++++++++++ config.example.json | 22 ++ config_test.json | 15 + src/webdrop_bridge/config.py | 199 +++++++++- src/webdrop_bridge/core/drag_interceptor.py | 158 +++++--- src/webdrop_bridge/core/url_converter.py | 86 ++++ src/webdrop_bridge/core/validator.py | 64 +-- src/webdrop_bridge/main.py | 8 +- src/webdrop_bridge/ui/bridge_script.js | 213 ++++++++-- src/webdrop_bridge/ui/download_interceptor.js | 72 ++++ src/webdrop_bridge/ui/main_window.py | 373 ++++++++++++++++-- src/webdrop_bridge/ui/restricted_web_view.py | 162 +++++++- test_download.html | 95 +++++ test_download.py | 19 + test_url_mappings.py | 24 ++ tests/unit/test_config.py | 7 +- tests/unit/test_drag_interceptor.py | 336 ++++++---------- tests/unit/test_main_window.py | 25 +- tests/unit/test_url_converter.py | 144 +++++++ tests/unit/test_validator.py | 7 +- webapp/index.html | 64 ++- 21 files changed, 1870 insertions(+), 432 deletions(-) create mode 100644 CONFIG_README.md create mode 100644 config.example.json create mode 100644 config_test.json create mode 100644 src/webdrop_bridge/core/url_converter.py create mode 100644 src/webdrop_bridge/ui/download_interceptor.js create mode 100644 test_download.html create mode 100644 test_download.py create mode 100644 test_url_mappings.py create mode 100644 tests/unit/test_url_converter.py diff --git a/CONFIG_README.md b/CONFIG_README.md new file mode 100644 index 0000000..261e7db --- /dev/null +++ b/CONFIG_README.md @@ -0,0 +1,209 @@ +# WebDrop Bridge Configuration Guide + +## Configuration File Location + +WebDrop Bridge supports two configuration methods: + +1. **JSON Configuration File** (Recommended for Azure URL mapping) + - Windows: `%APPDATA%\webdrop_bridge\config.json` + - macOS/Linux: `~/.config/webdrop_bridge/config.json` + +2. **Environment Variables** (`.env` file in project root) + - Used as fallback if JSON config doesn't exist + +## JSON Configuration Format + +Create a `config.json` file with the following structure: + +```json +{ + "app_name": "WebDrop Bridge", + "webapp_url": "https://wps.agravity.io/", + "url_mappings": [ + { + "url_prefix": "https://wpsagravitystg.file.core.windows.net/wpsagravitysync/", + "local_path": "Z:" + } + ], + "allowed_roots": [ + "Z:\\" + ], + "allowed_urls": [], + "check_file_exists": true, + "auto_check_updates": true, + "update_check_interval_hours": 24, + "log_level": "INFO", + "log_file": "logs/webdrop_bridge.log", + "window_width": 1024, + "window_height": 768, + "enable_logging": true +} +``` + +## Configuration Options + +### Core Settings + +- **`webapp_url`** (string): URL of the web application to load + - Example: `"https://wps.agravity.io/"` + - Supports `http://`, `https://`, or `file:///` URLs + +- **`url_mappings`** (array): Azure Blob Storage URL to local path mappings + - Each mapping has: + - `url_prefix`: Azure URL prefix (must end with `/`) + - `local_path`: Local drive letter or path + - Example: + ```json + { + "url_prefix": "https://wpsagravitystg.file.core.windows.net/wpsagravitysync/", + "local_path": "Z:" + } + ``` + +- **`allowed_roots`** (array): Whitelisted root directories for file access + - Security feature: Only files within these directories can be dragged + - Example: `["Z:\\", "C:\\Users\\Public"]` + +### Azure URL Mapping Example + +When the web application provides a drag URL like: +``` +https://wpsagravitystg.file.core.windows.net/wpsagravitysync/aN5PysnXIuRECzcRbvHkjL7g0/Hintergrund_Agravity.png +``` + +It will be converted to: +``` +Z:\aN5PysnXIuRECzcRbvHkjL7g0\Hintergrund_Agravity.png +``` + +### Security Settings + +- **`check_file_exists`** (boolean): Validate files exist before allowing drag + - Default: `true` + - Set to `false` only for testing + +- **`allowed_urls`** (array): Allowed URL patterns for web content + - Empty array = no restriction + - Example: `["wps.agravity.io", "*.example.com"]` + +### Update Settings + +- **`auto_check_updates`** (boolean): Automatically check for updates on startup + - Default: `true` + +- **`update_check_interval_hours`** (number): Hours between update checks + - Default: `24` + +### UI Settings + +- **`window_width`**, **`window_height`** (number): Initial window size in pixels + - Default: `1024` x `768` + +- **`log_level`** (string): Logging verbosity + - Options: `"DEBUG"`, `"INFO"`, `"WARNING"`, `"ERROR"`, `"CRITICAL"` + - Default: `"INFO"` + +- **`enable_logging`** (boolean): Whether to write logs to file + - Default: `true` + +## Quick Start + +1. Copy `config.example.json` to your config directory: + ```powershell + # Windows + mkdir "$env:APPDATA\webdrop_bridge" + copy config.example.json "$env:APPDATA\webdrop_bridge\config.json" + ``` + +2. Edit the configuration file with your Azure URL mappings and local paths + +3. Restart WebDrop Bridge + +## Multiple URL Mappings + +You can configure multiple Azure storage accounts: + +```json +{ + "url_mappings": [ + { + "url_prefix": "https://storage1.file.core.windows.net/container1/", + "local_path": "Z:" + }, + { + "url_prefix": "https://storage2.file.core.windows.net/container2/", + "local_path": "Y:" + } + ], + "allowed_roots": [ + "Z:\\", + "Y:\\" + ] +} +``` + +## Environment Variable Fallback + +If no JSON config exists, WebDrop Bridge will load from `.env`: + +```env +APP_NAME=WebDrop Bridge +WEBAPP_URL=https://wps.agravity.io/ +ALLOWED_ROOTS=Z:/ +LOG_LEVEL=INFO +WINDOW_WIDTH=1024 +WINDOW_HEIGHT=768 +``` + +**Note:** Environment variables don't support `url_mappings`. Use JSON config for Azure URL mapping. + +## Troubleshooting + +### "No mapping found for URL" +- Check that `url_prefix` matches the Azure URL exactly (including trailing `/`) +- Verify `url_mappings` is configured in your JSON config file +- Check logs in `logs/webdrop_bridge.log` + +### "Path is not within allowed roots" +- Ensure the mapped local path (e.g., `Z:`) is listed in `allowed_roots` +- Make sure the drive is mounted and accessible + +### "File does not exist" +- Verify the Azure sync is working and files are available locally +- Set `check_file_exists: false` temporarily for testing +- Check that the path mapping is correct + +## Example Configurations + +### Production (Agravity WPS) + +```json +{ + "webapp_url": "https://wps.agravity.io/", + "url_mappings": [ + { + "url_prefix": "https://wpsagravitystg.file.core.windows.net/wpsagravitysync/", + "local_path": "Z:" + } + ], + "allowed_roots": ["Z:\\"], + "log_level": "INFO" +} +``` + +### Development (Local Testing) + +```json +{ + "webapp_url": "file:///./webapp/index.html", + "url_mappings": [ + { + "url_prefix": "https://test.blob.core.windows.net/test/", + "local_path": "C:\\temp\\test" + } + ], + "allowed_roots": ["C:\\temp\\test"], + "check_file_exists": false, + "log_level": "DEBUG" +} +``` diff --git a/config.example.json b/config.example.json new file mode 100644 index 0000000..8da1b1b --- /dev/null +++ b/config.example.json @@ -0,0 +1,22 @@ +{ + "app_name": "WebDrop Bridge", + "webapp_url": "https://wps.agravity.io/", + "url_mappings": [ + { + "url_prefix": "https://wpsagravitystg.file.core.windows.net/wpsagravitysync/", + "local_path": "Z:" + } + ], + "allowed_roots": [ + "Z:\\" + ], + "allowed_urls": [], + "check_file_exists": true, + "auto_check_updates": true, + "update_check_interval_hours": 24, + "log_level": "INFO", + "log_file": "logs/webdrop_bridge.log", + "window_width": 1024, + "window_height": 768, + "enable_logging": true +} diff --git a/config_test.json b/config_test.json new file mode 100644 index 0000000..3efdfcf --- /dev/null +++ b/config_test.json @@ -0,0 +1,15 @@ +{ + "app_name": "WebDrop Bridge - Download Test", + "webapp_url": "test_download.html", + "url_mappings": [], + "allowed_roots": [], + "allowed_urls": [], + "check_file_exists": true, + "auto_check_updates": false, + "update_check_interval_hours": 24, + "log_level": "DEBUG", + "log_file": "logs/webdrop_bridge.log", + "window_width": 1024, + "window_height": 768, + "enable_logging": true +} diff --git a/src/webdrop_bridge/config.py b/src/webdrop_bridge/config.py index ea0c5a6..667bd04 100644 --- a/src/webdrop_bridge/config.py +++ b/src/webdrop_bridge/config.py @@ -1,8 +1,9 @@ """Configuration management for WebDrop Bridge application.""" +import json import logging import os -from dataclasses import dataclass +from dataclasses import dataclass, field from pathlib import Path from typing import List @@ -17,9 +18,29 @@ class ConfigurationError(Exception): pass +@dataclass +class URLMapping: + """Maps an Azure Blob Storage URL prefix to a local drive path.""" + + url_prefix: str + local_path: str + + def __post_init__(self): + """Validate mapping configuration.""" + if not self.url_prefix.startswith(("http://", "https://")): + raise ConfigurationError( + f"URL prefix must start with http:// or https://: {self.url_prefix}" + ) + # Ensure URL prefix ends with / + if not self.url_prefix.endswith("/"): + self.url_prefix += "/" + # Normalize local path + self.local_path = str(Path(self.local_path)) + + @dataclass class Config: - """Application configuration loaded from environment variables. + """Application configuration loaded from environment variables or JSON file. Attributes: app_name: Application display name @@ -28,7 +49,11 @@ class Config: log_file: Optional log file path allowed_roots: List of whitelisted root directories for file access allowed_urls: List of whitelisted URL domains/patterns (empty = no restriction) - webapp_url: URL to load in embedded web application + webapp_url: URL to load in embedded web application (default: https://wps.agravity.io/) + url_mappings: List of Azure URL to local path mappings + check_file_exists: Whether to validate that files exist before drag + auto_check_updates: Whether to automatically check for updates + update_check_interval_hours: Hours between update checks window_width: Initial window width in pixels window_height: Initial window height in pixels window_title: Main window title (default: "{app_name} v{app_version}") @@ -45,10 +70,85 @@ class Config: allowed_roots: List[Path] allowed_urls: List[str] webapp_url: str - window_width: int - window_height: int - window_title: str - enable_logging: bool + url_mappings: List[URLMapping] = field(default_factory=list) + check_file_exists: bool = True + auto_check_updates: bool = True + update_check_interval_hours: int = 24 + window_width: int = 1024 + window_height: int = 768 + window_title: str = "" + enable_logging: bool = True + + @classmethod + def from_file(cls, config_path: Path) -> "Config": + """Load configuration from JSON file. + + Args: + config_path: Path to configuration file + + Returns: + Config: Configured instance from JSON file + + Raises: + ConfigurationError: If configuration file is invalid + """ + if not config_path.exists(): + raise ConfigurationError(f"Configuration file not found: {config_path}") + + try: + with open(config_path, "r", encoding="utf-8") as f: + data = json.load(f) + except (json.JSONDecodeError, IOError) as e: + raise ConfigurationError(f"Failed to load configuration: {e}") from e + + # Get version from package + from webdrop_bridge import __version__ + + # Parse URL mappings + mappings = [ + URLMapping( + url_prefix=m["url_prefix"], + local_path=m["local_path"] + ) + for m in data.get("url_mappings", []) + ] + + # Parse allowed roots + allowed_roots = [Path(p).resolve() for p in data.get("allowed_roots", [])] + + # Validate allowed roots exist + for root in allowed_roots: + if not root.exists(): + logger.warning(f"Allowed root does not exist: {root}") + elif not root.is_dir(): + raise ConfigurationError(f"Allowed root is not a directory: {root}") + + # Get log file path + log_file = None + if data.get("enable_logging", True): + log_file_str = data.get("log_file", "logs/webdrop_bridge.log") + log_file = Path(log_file_str).resolve() + + app_name = data.get("app_name", "WebDrop Bridge") + window_title = data.get("window_title", f"{app_name} v{__version__}") + + return cls( + app_name=app_name, + app_version=__version__, + log_level=data.get("log_level", "INFO").upper(), + log_file=log_file, + allowed_roots=allowed_roots, + allowed_urls=data.get("allowed_urls", []), + webapp_url=data.get("webapp_url", "https://wps.agravity.io/"), + url_mappings=mappings, + check_file_exists=data.get("check_file_exists", True), + auto_check_updates=data.get("auto_check_updates", True), + update_check_interval_hours=data.get("update_check_interval_hours", 24), + window_width=data.get("window_width", 1024), + window_height=data.get("window_height", 768), + window_title=window_title, + enable_logging=data.get("enable_logging", True), + ) @classmethod def from_env(cls, env_file: str | None = None) -> "Config": @@ -81,7 +181,7 @@ class Config: log_file_str = os.getenv("LOG_FILE", "logs/webdrop_bridge.log") allowed_roots_str = os.getenv("ALLOWED_ROOTS", "Z:/,C:/Users/Public") allowed_urls_str = os.getenv("ALLOWED_URLS", "") - webapp_url = os.getenv("WEBAPP_URL", "file:///./webapp/index.html") + webapp_url = os.getenv("WEBAPP_URL", "https://wps.agravity.io/") window_width = int(os.getenv("WINDOW_WIDTH", "1024")) window_height = int(os.getenv("WINDOW_HEIGHT", "768")) # Window title defaults to app_name + version if not specified @@ -103,14 +203,13 @@ class Config: for p in allowed_roots_str.split(","): root_path = Path(p.strip()).resolve() if not root_path.exists(): - raise ConfigurationError( - f"Allowed root '{p.strip()}' does not exist" - ) - if not root_path.is_dir(): + logger.warning(f"Allowed root does not exist: {p.strip()}") + elif not root_path.is_dir(): raise ConfigurationError( f"Allowed root '{p.strip()}' is not a directory" ) - allowed_roots.append(root_path) + else: + allowed_roots.append(root_path) except ConfigurationError: raise except (ValueError, OSError) as e: @@ -140,6 +239,32 @@ class Config: if url.strip() ] if allowed_urls_str else [] + # Parse URL mappings (Azure Blob Storage โ†’ Local Paths) + # Format: url_prefix1=local_path1;url_prefix2=local_path2 + url_mappings_str = os.getenv("URL_MAPPINGS", "") + url_mappings = [] + if url_mappings_str: + try: + for mapping in url_mappings_str.split(";"): + mapping = mapping.strip() + if not mapping: + continue + if "=" not in mapping: + raise ConfigurationError( + f"Invalid URL mapping format: {mapping}. Expected 'url=path'" + ) + url_prefix, local_path_str = mapping.split("=", 1) + url_mappings.append( + URLMapping( + url_prefix=url_prefix.strip(), + local_path=local_path_str.strip() + ) + ) + except (ValueError, OSError) as e: + raise ConfigurationError( + f"Invalid URL_MAPPINGS: {url_mappings_str}. Error: {e}" + ) from e + return cls( app_name=app_name, app_version=app_version, @@ -148,12 +273,60 @@ class Config: allowed_roots=allowed_roots, allowed_urls=allowed_urls, webapp_url=webapp_url, + url_mappings=url_mappings, window_width=window_width, window_height=window_height, window_title=window_title, enable_logging=enable_logging, ) + def to_file(self, config_path: Path) -> None: + """Save configuration to JSON file. + + Args: + config_path: Path to save configuration + """ + data = { + "app_name": self.app_name, + "webapp_url": self.webapp_url, + "url_mappings": [ + { + "url_prefix": m.url_prefix, + "local_path": m.local_path + } + for m in self.url_mappings + ], + "allowed_roots": [str(p) for p in self.allowed_roots], + "allowed_urls": self.allowed_urls, + "check_file_exists": self.check_file_exists, + "auto_check_updates": self.auto_check_updates, + "update_check_interval_hours": self.update_check_interval_hours, + "log_level": self.log_level, + "log_file": str(self.log_file) if self.log_file else None, + "window_width": self.window_width, + "window_height": self.window_height, + "window_title": self.window_title, + "enable_logging": self.enable_logging, + } + + config_path.parent.mkdir(parents=True, exist_ok=True) + with open(config_path, "w", encoding="utf-8") as f: + json.dump(data, f, indent=2) + + @staticmethod + def get_default_config_path() -> Path: + """Get the default configuration file path. + + Returns: + Path to default config file + """ + import platform + if platform.system() == "Windows": + base = Path.home() / "AppData" / "Roaming" + else: + base = Path.home() / ".config" + return base / "webdrop_bridge" / "config.json" + def __repr__(self) -> str: """Return developer-friendly representation.""" return ( diff --git a/src/webdrop_bridge/core/drag_interceptor.py b/src/webdrop_bridge/core/drag_interceptor.py index 14502b5..f00eae2 100644 --- a/src/webdrop_bridge/core/drag_interceptor.py +++ b/src/webdrop_bridge/core/drag_interceptor.py @@ -1,5 +1,6 @@ """Qt widget for intercepting drag events and initiating native drag operations.""" +import logging from pathlib import Path from typing import List, Optional @@ -7,98 +8,129 @@ from PySide6.QtCore import QMimeData, Qt, QUrl, Signal from PySide6.QtGui import QDrag from PySide6.QtWidgets import QWidget +from webdrop_bridge.config import Config +from webdrop_bridge.core.url_converter import URLConverter from webdrop_bridge.core.validator import PathValidator, ValidationError +logger = logging.getLogger(__name__) + class DragInterceptor(QWidget): - """Widget that handles drag initiation for file paths. + """Widget that handles drag initiation for file paths or Azure URLs. - Intercepts drag events from web content and initiates native Qt drag - operations, allowing files to be dragged from web content to native - applications. + Intercepts drag events from web content, converts Azure Blob Storage URLs + to local paths, validates them, and initiates native Qt drag operations. Signals: drag_started: Emitted when a drag operation begins successfully drag_failed: Emitted when drag initiation fails """ - # Signals with string parameters (file paths that were dragged) - drag_started = Signal(list) # List[str] - list of file paths - drag_failed = Signal(str) # str - error message + # Signals with string parameters + drag_started = Signal(str, str) # (url_or_path, local_path) + drag_failed = Signal(str, str) # (url_or_path, error_message) - def __init__(self, parent: Optional[QWidget] = None): + def __init__(self, config: Config, parent: Optional[QWidget] = None): """Initialize the drag interceptor. Args: + config: Application configuration parent: Parent widget """ super().__init__(parent) - self._validator: Optional[PathValidator] = None + self.config = config + self._validator = PathValidator( + config.allowed_roots, + check_file_exists=config.check_file_exists + ) + self._url_converter = URLConverter(config) - def set_validator(self, validator: PathValidator) -> None: - """Set the path validator for this interceptor. + def handle_drag(self, text: str) -> bool: + """Handle drag event from web view. + + Determines if the text is an Azure URL or file path, converts if needed, + validates, and initiates native drag operation. Args: - validator: PathValidator instance to use for validation - """ - self._validator = validator - - def initiate_drag(self, file_paths: List[str]) -> bool: - """Initiate a native drag operation for the given files. - - Args: - file_paths: List of file paths to drag + text: Azure Blob Storage URL or file path from web drag Returns: - True if drag was successfully initiated, False otherwise + True if native drag was initiated, False otherwise """ - if not file_paths: - self.drag_failed.emit("No files to drag") + if not text or not text.strip(): + error_msg = "Empty drag text" + logger.warning(error_msg) + self.drag_failed.emit("", error_msg) return False - if not self._validator: - self.drag_failed.emit("Validator not configured") - return False + text = text.strip() + logger.debug(f"Handling drag for text: {text}") - # Validate all paths first - validated_paths = [] - for path_str in file_paths: - try: - path = Path(path_str) - if self._validator.validate(path): - validated_paths.append(path) - except ValidationError as e: - self.drag_failed.emit(f"Validation failed for {path_str}: {e}") + # Check if it's an Azure URL and convert to local path + if self._url_converter.is_azure_url(text): + local_path = self._url_converter.convert_url_to_path(text) + if local_path is None: + error_msg = "No mapping found for URL" + logger.warning(f"{error_msg}: {text}") + self.drag_failed.emit(text, error_msg) return False - - if not validated_paths: - self.drag_failed.emit("No valid files after validation") - return False - - # Create MIME data with file URLs - mime_data = QMimeData() - file_urls = [ - path.as_uri() for path in validated_paths - ] - mime_data.setUrls([QUrl(url) for url in file_urls]) - - # Create and execute drag operation - drag = QDrag(self) - drag.setMimeData(mime_data) - # Use default drag pixmap (small icon) - drag.setPixmap(self.grab(self.rect()).scaled( - 64, 64, Qt.AspectRatioMode.KeepAspectRatio - )) - - # Execute drag operation (blocking call) - drop_action = drag.exec(Qt.DropAction.CopyAction) - - # Check result - if drop_action == Qt.DropAction.CopyAction: - self.drag_started.emit(validated_paths) - return True + source_text = text else: - self.drag_failed.emit("Drag operation cancelled or failed") + # Treat as direct file path + local_path = Path(text) + source_text = text + + # Validate the path + try: + self._validator.validate(local_path) + except ValidationError as e: + error_msg = str(e) + logger.warning(f"Validation failed for {local_path}: {error_msg}") + self.drag_failed.emit(source_text, error_msg) return False + logger.info(f"Initiating drag for: {local_path}") + # Create native file drag + success = self._create_native_drag(local_path) + + if success: + self.drag_started.emit(source_text, str(local_path)) + else: + error_msg = "Failed to create native drag operation" + logger.error(error_msg) + self.drag_failed.emit(source_text, error_msg) + + return success + + def _create_native_drag(self, file_path: Path) -> bool: + """Create a native file system drag operation. + + Args: + file_path: Local file path to drag + + Returns: + True if drag was created successfully + """ + try: + # Create MIME data with file URL + mime_data = QMimeData() + file_url = QUrl.fromLocalFile(str(file_path)) + mime_data.setUrls([file_url]) + + # Create and execute drag + drag = QDrag(self) + drag.setMimeData(mime_data) + + # Optional: Set a drag icon/pixmap if available + # drag.setPixmap(...) + + # Start drag operation (blocks until drop or cancel) + # Qt.CopyAction allows copying files + result = drag.exec(Qt.DropAction.CopyAction) + + return result == Qt.DropAction.CopyAction + + except Exception as e: + logger.exception(f"Error creating native drag: {e}") + return False diff --git a/src/webdrop_bridge/core/url_converter.py b/src/webdrop_bridge/core/url_converter.py new file mode 100644 index 0000000..4841dfe --- /dev/null +++ b/src/webdrop_bridge/core/url_converter.py @@ -0,0 +1,86 @@ +"""URL to local path conversion for Azure Blob Storage URLs.""" + +import logging +from pathlib import Path +from typing import Optional +from urllib.parse import unquote + +from ..config import Config, URLMapping + +logger = logging.getLogger(__name__) + + +class URLConverter: + """Converts Azure Blob Storage URLs to local file paths.""" + + def __init__(self, config: Config): + """Initialize converter with configuration. + + Args: + config: Application configuration with URL mappings + """ + self.config = config + + def convert_url_to_path(self, url: str) -> Optional[Path]: + """Convert Azure Blob Storage URL to local file path. + + Args: + url: Azure Blob Storage URL + + Returns: + Local file path if mapping found, None otherwise + + Example: + >>> converter.convert_url_to_path( + ... "https://wpsagravitystg.file.core.windows.net/wpsagravitysync/aN5PysnXIuRECzcRbvHkjL7g0/file.png" + ... ) + Path("Z:/aN5PysnXIuRECzcRbvHkjL7g0/file.png") + """ + if not url: + return None + + # URL decode (handles special characters like spaces) + url = unquote(url) + + # Find matching URL mapping + for mapping in self.config.url_mappings: + if url.startswith(mapping.url_prefix): + # Extract relative path after prefix + relative_path = url[len(mapping.url_prefix):] + + # Combine with local path + local_path = Path(mapping.local_path) / relative_path + + # Normalize path (resolve .. and .) but don't follow symlinks yet + try: + # On Windows, normalize separators + local_path = Path(str(local_path).replace("/", "\\")) + except (OSError, RuntimeError) as e: + logger.warning(f"Failed to normalize path {local_path}: {e}") + return None + + logger.debug(f"Converted URL to path: {url} -> {local_path}") + return local_path + + logger.debug(f"No mapping found for URL: {url}") + return None + + def is_azure_url(self, text: str) -> bool: + """Check if text is an Azure Blob Storage URL. + + Args: + text: Text to check + + Returns: + True if text matches configured URL prefixes + """ + if not text: + return False + + text = text.strip() + + for mapping in self.config.url_mappings: + if text.startswith(mapping.url_prefix): + return True + + return False diff --git a/src/webdrop_bridge/core/validator.py b/src/webdrop_bridge/core/validator.py index 218fe2d..aac28c0 100644 --- a/src/webdrop_bridge/core/validator.py +++ b/src/webdrop_bridge/core/validator.py @@ -1,7 +1,10 @@ """Path validation for secure file operations.""" +import logging from pathlib import Path -from typing import List +from typing import List, Optional + +logger = logging.getLogger(__name__) class ValidationError(Exception): @@ -18,28 +21,27 @@ class PathValidator: directory traversal attacks. """ - def __init__(self, allowed_roots: List[Path]): + def __init__(self, allowed_roots: List[Path], check_file_exists: bool = True): """Initialize validator with allowed root directories. Args: allowed_roots: List of Path objects representing allowed root dirs + check_file_exists: Whether to validate that files exist Raises: ValidationError: If any root doesn't exist or isn't a directory """ self.allowed_roots = [] + self.check_file_exists = check_file_exists for root in allowed_roots: root_path = Path(root).resolve() if not root_path.exists(): - raise ValidationError( - f"Allowed root '{root}' does not exist" - ) - if not root_path.is_dir(): - raise ValidationError( - f"Allowed root '{root}' is not a directory" - ) - self.allowed_roots.append(root_path) + logger.warning(f"Allowed root '{root}' does not exist") + elif not root_path.is_dir(): + raise ValidationError(f"Allowed root '{root}' is not a directory") + else: + self.allowed_roots.append(root_path) def validate(self, path: Path) -> bool: """Validate that path is within an allowed root directory. @@ -59,28 +61,32 @@ class PathValidator: except (OSError, ValueError) as e: raise ValidationError(f"Cannot resolve path '{path}': {e}") from e - # Check file exists - if not file_path.exists(): - raise ValidationError(f"File does not exist: {path}") + # Check file exists if required + if self.check_file_exists: + if not file_path.exists(): + raise ValidationError(f"File does not exist: {path}") - # Check it's a regular file (not directory, symlink to dir, etc) - if not file_path.is_file(): - raise ValidationError(f"Path is not a regular file: {path}") + # Check it's a regular file (not directory, symlink to dir, etc) + if not file_path.is_file(): + raise ValidationError(f"Path is not a regular file: {path}") - # Check path is within an allowed root - for allowed_root in self.allowed_roots: - try: - # This raises ValueError if file_path is not relative to root - file_path.relative_to(allowed_root) - return True - except ValueError: - continue + # Check path is within an allowed root (if roots configured) + if self.allowed_roots: + for allowed_root in self.allowed_roots: + try: + # This raises ValueError if file_path is not relative to root + file_path.relative_to(allowed_root) + return True + except ValueError: + continue - # Not in any allowed root - raise ValidationError( - f"Path '{file_path}' is not within allowed roots: " - f"{self.allowed_roots}" - ) + # Not in any allowed root + raise ValidationError( + f"Path '{file_path}' is not within allowed roots: " + f"{self.allowed_roots}" + ) + + return True def is_valid(self, path: Path) -> bool: """Check if path is valid without raising exception. diff --git a/src/webdrop_bridge/main.py b/src/webdrop_bridge/main.py index 6c33e88..d63ad97 100644 --- a/src/webdrop_bridge/main.py +++ b/src/webdrop_bridge/main.py @@ -19,8 +19,12 @@ def main() -> int: int: Exit code (0 for success, non-zero for error) """ try: - # Load configuration from environment - config = Config.from_env() + # Load configuration from file if it exists, otherwise from environment + config_path = Config.get_default_config_path() + if config_path.exists(): + config = Config.from_file(config_path) + else: + config = Config.from_env() # Set up logging log_file = None diff --git a/src/webdrop_bridge/ui/bridge_script.js b/src/webdrop_bridge/ui/bridge_script.js index aa5b8a3..622a0e4 100644 --- a/src/webdrop_bridge/ui/bridge_script.js +++ b/src/webdrop_bridge/ui/bridge_script.js @@ -5,6 +5,58 @@ if (window.__webdrop_bridge_injected) return; window.__webdrop_bridge_injected = true; + console.log('[WebDrop Bridge] Script loaded'); + + // Store web app's dragstart handlers by intercepting addEventListener + var webAppDragHandlers = []; + var originalAddEventListener = EventTarget.prototype.addEventListener; + var listenerPatchActive = true; + + // Patch addEventListener to intercept dragstart registrations + EventTarget.prototype.addEventListener = function(type, listener, options) { + if (listenerPatchActive && type === 'dragstart' && listener) { + // Store the web app's dragstart handler instead of registering it + console.log('[WebDrop Bridge] Intercepted dragstart listener registration on', this.tagName || this.constructor.name); + webAppDragHandlers.push({ + target: this, + listener: listener, + options: options + }); + return; + } + // All other events: use original + return originalAddEventListener.call(this, type, listener, options); + }; + + // Patch DataTransfer.setData to intercept URL setting by the web app + var originalSetData = null; + var currentDragData = null; + + try { + if (DataTransfer.prototype.setData) { + originalSetData = DataTransfer.prototype.setData; + + DataTransfer.prototype.setData = function(format, data) { + // Store the data for our analysis + if (format === 'text/plain' || format === 'text/uri-list') { + currentDragData = data; + console.log('[WebDrop Bridge] DataTransfer.setData intercepted:', format, '=', data.substring(0, 80)); + + // Log via bridge if available + if (window.bridge && typeof window.bridge.debug_log === 'function') { + window.bridge.debug_log('setData intercepted: ' + format + ' = ' + data.substring(0, 60)); + } + } + // Call original to maintain web app functionality + return originalSetData.call(this, format, data); + }; + + console.log('[WebDrop Bridge] DataTransfer.setData patched'); + } + } catch(e) { + console.error('[WebDrop Bridge] Failed to patch DataTransfer:', e); + } + function ensureChannel(cb) { if (window.bridge) { cb(); return; } @@ -12,62 +64,151 @@ if (window.QWebChannel && window.qt && window.qt.webChannelTransport) { new QWebChannel(window.qt.webChannelTransport, function(channel) { window.bridge = channel.objects.bridge; + console.log('[WebDrop Bridge] QWebChannel connected'); cb(); }); + } else { + // If QWebChannel is not available, log error + console.error('[WebDrop Bridge] QWebChannel not available! Check if qwebchannel.js was loaded.'); } } + // QWebChannel should already be loaded inline (no need to load from qrc://) if (window.QWebChannel) { init(); - return; + } else { + console.error('[WebDrop Bridge] QWebChannel not found! Cannot initialize bridge.'); } - - var s = document.createElement('script'); - s.src = 'qrc:///qtwebchannel/qwebchannel.js'; - s.onload = init; - document.documentElement.appendChild(s); } function hook() { - document.addEventListener('dragstart', function(e) { - var dt = e.dataTransfer; - if (!dt) return; - - // Get path from existing payload or from the card markup. - var path = dt.getData('text/plain'); - if (!path) { - var card = e.target.closest && e.target.closest('.drag-item'); - if (card) { - var pathEl = card.querySelector('p'); - if (pathEl) { - path = (pathEl.textContent || '').trim(); + console.log('[WebDrop Bridge] Installing hook, have ' + webAppDragHandlers.length + ' intercepted handlers'); + + if (window.bridge && typeof window.bridge.debug_log === 'function') { + window.bridge.debug_log('Installing drag interceptor with ' + webAppDragHandlers.length + ' intercepted handlers'); + } + + // Stop intercepting addEventListener - from now on, listeners register normally + listenerPatchActive = false; + + // Register OUR dragstart handler using capture phase on document + originalAddEventListener.call(document, 'dragstart', function(e) { + try { + console.log('[WebDrop Bridge] >>> DRAGSTART fired on:', e.target.tagName, 'altKey:', e.altKey, 'currentDragData:', currentDragData); + + if (window.bridge && typeof window.bridge.debug_log === 'function') { + window.bridge.debug_log('dragstart fired on ' + e.target.tagName + ' altKey=' + e.altKey); + } + + // Only intercept if ALT key is pressed (web app's text drag mode) + if (!e.altKey) { + console.log('[WebDrop Bridge] ALT not pressed, ignoring drag (normal web app drag)'); + return; // Let web app handle normal drags + } + + console.log('[WebDrop Bridge] ALT pressed - processing for file drag conversion'); + + // Manually invoke all the web app's dragstart handlers + var handlersInvoked = 0; + console.log('[WebDrop Bridge] About to invoke', webAppDragHandlers.length, 'stored handlers'); + + for (var i = 0; i < webAppDragHandlers.length; i++) { + try { + var handler = webAppDragHandlers[i]; + // Check if this handler should be called for this target + if (handler.target === document || + handler.target === e.target || + (handler.target.contains && handler.target.contains(e.target))) { + + console.log('[WebDrop Bridge] Calling stored handler #' + i); + handler.listener.call(e.target, e); + handlersInvoked++; + } + } catch (err) { + console.error('[WebDrop Bridge] Error calling web app handler #' + i + ':', err); } } + + console.log('[WebDrop Bridge] Invoked', handlersInvoked, 'handlers, currentDragData:', currentDragData ? currentDragData.substring(0, 60) : 'null'); + + // NOW check if we have a convertible URL + if (currentDragData) { + console.log('[WebDrop Bridge] Checking currentDragData:', currentDragData.substring(0, 80)); + var path = currentDragData; + var isZDrive = /^z:/i.test(path); + var isAzureUrl = /^https?:\/\/.+\.file\.core\.windows\.net\//i.test(path); + + console.log('[WebDrop Bridge] isZDrive:', isZDrive, 'isAzureUrl:', isAzureUrl); + + if (isZDrive || isAzureUrl) { + console.log('[WebDrop Bridge] >>> CONVERTING URL TO NATIVE DRAG'); + + if (window.bridge && typeof window.bridge.debug_log === 'function') { + window.bridge.debug_log('Convertible URL detected - preventing browser drag'); + } + + // Prevent the browser's drag operation + e.preventDefault(); + e.stopPropagation(); + + // Start native file drag via Qt + ensureChannel(function() { + if (window.bridge && typeof window.bridge.start_file_drag === 'function') { + console.log('[WebDrop Bridge] Calling start_file_drag:', path.substring(0, 60)); + window.bridge.start_file_drag(path); + currentDragData = null; + } else { + console.error('[WebDrop Bridge] bridge.start_file_drag not available!'); + } + }); + + return false; + } else { + console.log('[WebDrop Bridge] URL not convertible:', path.substring(0, 60)); + } + } else { + console.log('[WebDrop Bridge] No currentDragData set'); + } + } catch (mainError) { + console.error('[WebDrop Bridge] CRITICAL ERROR in dragstart handler:', mainError); + if (window.bridge && typeof window.bridge.debug_log === 'function') { + window.bridge.debug_log('ERROR in dragstart: ' + mainError.message); + } } - if (!path) return; + }, true); // CAPTURE PHASE - intercept early + + // Reset state on dragend + originalAddEventListener.call(document, 'dragend', function(e) { + currentDragData = null; + }, false); + + console.log('[WebDrop Bridge] Drag listener registered on document (capture phase)'); + } - // Ensure text payload exists for non-file drags and downstream targets. - if (!dt.getData('text/plain')) { - dt.setData('text/plain', path); - } - - // Check if path is Z:\ โ€” if yes, trigger native file drag. Otherwise, stay as text. - var isZDrive = /^z:/i.test(path); - if (!isZDrive) return; - - // Z:\ detected โ€” prevent default browser drag and convert to native file drag - e.preventDefault(); + // Wait for DOMContentLoaded and then a bit more before installing hook + // This gives the web app time to register its handlers + function installHook() { + console.log('[WebDrop Bridge] DOM ready, waiting 2 seconds for web app to register handlers...'); + console.log('[WebDrop Bridge] Currently have', webAppDragHandlers.length, 'intercepted handlers'); + + setTimeout(function() { + console.log('[WebDrop Bridge] Installing hook now, have', webAppDragHandlers.length, 'intercepted handlers'); + hook(); + ensureChannel(function() { - if (window.bridge && typeof window.bridge.start_file_drag === 'function') { - window.bridge.start_file_drag(path); + if (window.bridge && typeof window.bridge.debug_log === 'function') { + window.bridge.debug_log('Hook installed with ' + webAppDragHandlers.length + ' captured handlers'); } }); - }, false); + }, 2000); // Wait 2 seconds after DOM ready } - + + // Install after DOM is ready if (document.readyState === 'loading') { - document.addEventListener('DOMContentLoaded', hook); + console.log('[WebDrop Bridge] Waiting for DOMContentLoaded...'); + originalAddEventListener.call(document, 'DOMContentLoaded', installHook); } else { - hook(); + console.log('[WebDrop Bridge] DOM already ready, installing hook...'); + installHook(); } })(); diff --git a/src/webdrop_bridge/ui/download_interceptor.js b/src/webdrop_bridge/ui/download_interceptor.js new file mode 100644 index 0000000..45a757e --- /dev/null +++ b/src/webdrop_bridge/ui/download_interceptor.js @@ -0,0 +1,72 @@ +// Download Interceptor Script +// Intercepts JavaScript-based downloads (fetch, XMLHttpRequest, Blob URLs) + +(function() { + 'use strict'; + + console.log('๐Ÿ” Download interceptor script loaded'); + + // Intercept fetch() calls + const originalFetch = window.fetch; + window.fetch = function(...args) { + const url = args[0]; + console.log('๐ŸŒ Fetch called:', url); + + // Check if this looks like a download + if (typeof url === 'string') { + const urlLower = url.toLowerCase(); + const downloadPatterns = [ + '/download', '/export', '/file', + '.pdf', '.zip', '.xlsx', '.docx', + 'attachment', 'content-disposition' + ]; + + if (downloadPatterns.some(pattern => urlLower.includes(pattern))) { + console.log('๐Ÿ“ฅ Potential download detected via fetch:', url); + } + } + + return originalFetch.apply(this, args); + }; + + // Intercept XMLHttpRequest + const originalXHROpen = XMLHttpRequest.prototype.open; + XMLHttpRequest.prototype.open = function(method, url, ...rest) { + console.log('๐ŸŒ XHR opened:', method, url); + this._url = url; + return originalXHROpen.apply(this, [method, url, ...rest]); + }; + + const originalXHRSend = XMLHttpRequest.prototype.send; + XMLHttpRequest.prototype.send = function(...args) { + console.log('๐Ÿ“ค XHR send:', this._url); + return originalXHRSend.apply(this, args); + }; + + // Intercept Blob URL creation + const originalCreateObjectURL = URL.createObjectURL; + URL.createObjectURL = function(blob) { + console.log('๐Ÿ”— Blob URL created, size:', blob.size, 'type:', blob.type); + return originalCreateObjectURL.apply(this, arguments); + }; + + // Intercept anchor clicks that might be downloads + document.addEventListener('click', function(e) { + const target = e.target.closest('a'); + if (target && target.href) { + const href = target.href; + const download = target.getAttribute('download'); + + if (download !== null) { + console.log('๐Ÿ“ฅ Download link clicked:', href, 'filename:', download); + } + + // Check for blob URLs + if (href.startsWith('blob:')) { + console.log('๐Ÿ“ฆ Blob download link clicked:', href); + } + } + }, true); + + console.log('โœ… Download interceptor active'); +})(); diff --git a/src/webdrop_bridge/ui/main_window.py b/src/webdrop_bridge/ui/main_window.py index 84638a4..252e27b 100644 --- a/src/webdrop_bridge/ui/main_window.py +++ b/src/webdrop_bridge/ui/main_window.py @@ -6,10 +6,22 @@ from datetime import datetime from pathlib import Path from typing import Optional -from PySide6.QtCore import QObject, QPoint, QSize, Qt, QThread, QTimer, QUrl, Signal, Slot -from PySide6.QtGui import QIcon +from PySide6.QtCore import ( + QEvent, + QObject, + QPoint, + QSize, + QStandardPaths, + Qt, + QThread, + QTimer, + QUrl, + Signal, + Slot, +) +from PySide6.QtGui import QIcon, QMouseEvent from PySide6.QtWebChannel import QWebChannel -from PySide6.QtWebEngineCore import QWebEngineScript +from PySide6.QtWebEngineCore import QWebEngineDownloadRequest, QWebEngineScript from PySide6.QtWidgets import ( QLabel, QMainWindow, @@ -202,19 +214,29 @@ class _DragBridge(QObject): @Slot(str) def start_file_drag(self, path_text: str) -> None: - """Start a native file drag for the given path. + """Start a native file drag for the given path or Azure URL. - Called from JavaScript when user drags a Z:\ path item. + Called from JavaScript when user drags an item. + Accepts either local file paths or Azure Blob Storage URLs. Defers execution to avoid Qt drag manager state issues. Args: - path_text: File path string to drag + path_text: File path string or Azure URL to drag """ logger.debug(f"Bridge: start_file_drag called for {path_text}") # Defer to avoid drag manager state issues - # initiate_drag() handles validation internally - QTimer.singleShot(0, lambda: self.window.drag_interceptor.initiate_drag([path_text])) + # handle_drag() handles URL conversion and validation internally + QTimer.singleShot(0, lambda: self.window.drag_interceptor.handle_drag(path_text)) + + @Slot(str) + def debug_log(self, message: str) -> None: + """Log debug message from JavaScript. + + Args: + message: Debug message from JavaScript + """ + logger.info(f"JS Debug: {message}") class MainWindow(QMainWindow): @@ -257,6 +279,16 @@ class MainWindow(QMainWindow): # Create web engine view self.web_view = RestrictedWebEngineView(config.allowed_urls) + + # Enable the main window and web view to receive drag events + self.setAcceptDrops(True) + self.web_view.setAcceptDrops(True) + + # Track ongoing drags from web view + self._current_drag_url = None + + # Redirect JavaScript console messages to Python logger + self.web_view.page().javaScriptConsoleMessage = self._on_js_console_message # Create navigation toolbar (Kiosk-mode navigation) self._create_navigation_toolbar() @@ -264,11 +296,8 @@ class MainWindow(QMainWindow): # Create status bar self._create_status_bar() - # Create drag interceptor - self.drag_interceptor = DragInterceptor() - # Set up path validator - validator = PathValidator(config.allowed_roots) - self.drag_interceptor.set_validator(validator) + # Create drag interceptor with config (includes URL converter) + self.drag_interceptor = DragInterceptor(config) # Connect drag interceptor signals self.drag_interceptor.drag_started.connect(self._on_drag_started) @@ -282,6 +311,26 @@ class MainWindow(QMainWindow): # Install the drag bridge script self._install_bridge_script() + + # Connect to loadFinished to verify script injection + self.web_view.loadFinished.connect(self._on_page_loaded) + + # Set up download handler + profile = self.web_view.page().profile() + logger.info(f"Connecting download handler to profile: {profile}") + + # CRITICAL: Connect download handler BEFORE any page loads + profile.downloadRequested.connect(self._on_download_requested) + + # Enable downloads by setting download path + downloads_path = QStandardPaths.writableLocation( + QStandardPaths.StandardLocation.DownloadLocation + ) + if downloads_path: + profile.setDownloadPath(downloads_path) + logger.info(f"Download path set to: {downloads_path}") + + logger.info("Download handler connected successfully") # Set up central widget with layout central_widget = QWidget() @@ -353,19 +402,55 @@ class MainWindow(QMainWindow): def _install_bridge_script(self) -> None: """Install the drag bridge JavaScript via QWebEngineScript. - Follows the POC pattern for proper script injection and QWebChannel setup. + Uses DocumentCreation injection point to ensure script runs as early as possible, + before any page scripts that might interfere with drag events. + + Embeds qwebchannel.js inline to avoid CSP issues with qrc:// URLs. """ + from PySide6.QtCore import QFile, QIODevice + script = QWebEngineScript() script.setName("webdrop-bridge") - script.setInjectionPoint(QWebEngineScript.InjectionPoint.DocumentReady) + script.setInjectionPoint(QWebEngineScript.InjectionPoint.DocumentCreation) script.setWorldId(QWebEngineScript.ScriptWorldId.MainWorld) script.setRunsOnSubFrames(False) + # Load qwebchannel.js from Qt resources (avoids CSP blocking qrc:// URLs) + qwebchannel_code = "" + qwebchannel_file = QFile(":/qtwebchannel/qwebchannel.js") + if qwebchannel_file.open(QIODevice.OpenModeFlag.ReadOnly | QIODevice.OpenModeFlag.Text): + qwebchannel_code = bytes(qwebchannel_file.readAll()).decode('utf-8') + qwebchannel_file.close() + logger.debug("Loaded qwebchannel.js inline to avoid CSP issues") + else: + logger.warning("Failed to load qwebchannel.js from resources") + # Load bridge script from file script_path = Path(__file__).parent / "bridge_script.js" try: with open(script_path, 'r', encoding='utf-8') as f: - script.setSourceCode(f.read()) + bridge_code = f.read() + + # Load download interceptor + download_interceptor_path = Path(__file__).parent / "download_interceptor.js" + download_interceptor_code = "" + try: + with open(download_interceptor_path, 'r', encoding='utf-8') as f: + download_interceptor_code = f.read() + logger.debug(f"Loaded download interceptor from {download_interceptor_path}") + except (OSError, IOError) as e: + logger.warning(f"Download interceptor not found: {e}") + + # Combine qwebchannel.js + bridge script + download interceptor (inline to avoid CSP) + if qwebchannel_code: + combined_code = qwebchannel_code + "\n\n" + bridge_code + else: + combined_code = bridge_code + + if download_interceptor_code: + combined_code += "\n\n" + download_interceptor_code + + script.setSourceCode(combined_code) self.web_view.page().scripts().insert(script) logger.debug(f"Installed bridge script from {script_path}") except (OSError, IOError) as e: @@ -399,23 +484,248 @@ class MainWindow(QMainWindow): # Silently fail if stylesheet can't be read pass - def _on_drag_started(self, paths: list) -> None: + def _on_drag_started(self, source: str, local_path: str) -> None: """Handle successful drag initiation. Args: - paths: List of paths that were dragged + source: Original URL or path from web content + local_path: Local file path that is being dragged """ - # Can be extended with logging or status bar updates - pass + logger.info(f"Drag started: {source} -> {local_path}") + # Can be extended with status bar updates or user feedback - def _on_drag_failed(self, error: str) -> None: + def _on_drag_failed(self, source: str, error: str) -> None: """Handle drag operation failure. Args: + source: Original URL or path from web content error: Error message """ - # Can be extended with logging or user notification - pass + logger.warning(f"Drag failed for {source}: {error}") + # Can be extended with user notification or status bar message + + def _on_download_requested(self, download: QWebEngineDownloadRequest) -> None: + """Handle download requests from the embedded web view. + + Downloads are automatically saved to the user's Downloads folder. + + Args: + download: Download request from the web engine + """ + logger.info("=" * 60) + logger.info("๐Ÿ”ฅ DOWNLOAD REQUESTED - Handler called!") + logger.info("=" * 60) + + try: + # Log all download details for debugging + logger.info(f"Download URL: {download.url().toString()}") + logger.info(f"Download filename: {download.downloadFileName()}") + logger.info(f"Download mime type: {download.mimeType()}") + logger.info(f"Download suggested filename: {download.suggestedFileName()}") + logger.info(f"Download state: {download.state()}") + + # Get the system's Downloads folder + downloads_path = QStandardPaths.writableLocation( + QStandardPaths.StandardLocation.DownloadLocation + ) + + if not downloads_path: + # Fallback to user's home directory if Downloads folder not available + downloads_path = str(Path.home()) + logger.warning("Downloads folder not found, using home directory") + + # Use suggested filename if available, fallback to downloadFileName + filename = download.suggestedFileName() or download.downloadFileName() + if not filename: + filename = "download" + logger.warning("No filename suggested, using 'download'") + + # Construct full download path + download_file = Path(downloads_path) / filename + logger.info(f"๐Ÿ“ Download will be saved to: {download_file}") + + # Set download path and accept + download.setDownloadDirectory(str(download_file.parent)) + download.setDownloadFileName(download_file.name) + download.accept() + + logger.info(f"โœ… Download accepted and started: {download_file}") + + # Update status bar (temporarily) + self.status_bar.showMessage( + f"๐Ÿ“ฅ Download: {filename}", 3000 + ) + + # Connect to state changed for progress tracking + download.stateChanged.connect( + lambda state: logger.info(f"Download state changed to: {state}") + ) + + # Connect to finished signal for completion feedback + download.isFinishedChanged.connect( + lambda: self._on_download_finished(download, download_file) + ) + + except Exception as e: + logger.error(f"โŒ Error handling download: {e}", exc_info=True) + self.status_bar.showMessage(f"โŒ Download-Fehler: {e}", 5000) + + def _on_download_finished(self, download: QWebEngineDownloadRequest, file_path: Path) -> None: + """Handle download completion. + + Args: + download: The completed download request + file_path: Path where file was saved + """ + try: + if not download.isFinished(): + return + + state = download.state() + logger.info(f"Download finished with state: {state}") + + if state == QWebEngineDownloadRequest.DownloadState.DownloadCompleted: + logger.info(f"Download completed successfully: {file_path}") + self.status_bar.showMessage( + f"โœ… Download abgeschlossen: {file_path.name}", 5000 + ) + elif state == QWebEngineDownloadRequest.DownloadState.DownloadCancelled: + logger.info(f"Download cancelled: {file_path.name}") + self.status_bar.showMessage( + f"โš ๏ธ Download abgebrochen: {file_path.name}", 3000 + ) + elif state == QWebEngineDownloadRequest.DownloadState.DownloadInterrupted: + logger.warning(f"Download interrupted: {file_path.name}") + self.status_bar.showMessage( + f"โŒ Download fehlgeschlagen: {file_path.name}", 5000 + ) + except Exception as e: + logger.error(f"Error in download finished handler: {e}", exc_info=True) + + def dragEnterEvent(self, event): + """Handle drag entering the main window (from WebView or external). + + When a drag from the WebView enters the MainWindow area, we can read + the drag data and potentially convert Azure URLs to file drags. + + Args: + event: QDragEnterEvent + """ + from PySide6.QtCore import QMimeData + + mime_data = event.mimeData() + + # Check if we have text data (URL from web app) + if mime_data.hasText(): + url_text = mime_data.text() + logger.debug(f"Drag entered main window with text: {url_text[:100]}") + + # Store for potential conversion + self._current_drag_url = url_text + + # Check if it's convertible + is_azure = url_text.startswith('https://') and 'file.core.windows.net' in url_text + is_z_drive = url_text.lower().startswith('z:') + + if is_azure or is_z_drive: + logger.info(f"Convertible URL detected in drag: {url_text[:60]}") + event.acceptProposedAction() + return + + event.ignore() + + def dragMoveEvent(self, event): + """Handle drag moving over the main window. + + Args: + event: QDragMoveEvent + """ + if self._current_drag_url: + event.acceptProposedAction() + else: + event.ignore() + + def dragLeaveEvent(self, event): + """Handle drag leaving the main window. + + Args: + event: QDragLeaveEvent + """ + logger.debug("Drag left main window") + # Reset tracking + self._current_drag_url = None + + def dropEvent(self, event): + """Handle drop on the main window. + + This captures drops on the MainWindow area (outside WebView). + If the user drops an Azure URL here, we convert it to a file operation. + + Args: + event: QDropEvent + """ + if self._current_drag_url: + logger.info(f"Drop on main window with URL: {self._current_drag_url[:60]}") + + # Handle via drag interceptor (converts Azure URL to local path) + success = self.drag_interceptor.handle_drag(self._current_drag_url) + + if success: + event.acceptProposedAction() + else: + event.ignore() + + self._current_drag_url = None + else: + event.ignore() + + def _on_js_console_message(self, level, message, line_number, source_id): + """Redirect JavaScript console messages to Python logger. + + Args: + level: Console message level (JavaScriptConsoleMessageLevel enum) + message: The console message + line_number: Line number where the message originated + source_id: Source file/URL where the message originated + """ + from PySide6.QtWebEngineCore import QWebEnginePage + + # Map JS log levels to Python log levels using enum + if level == QWebEnginePage.JavaScriptConsoleMessageLevel.InfoMessageLevel: + logger.info(f"JS Console: {message}") + elif level == QWebEnginePage.JavaScriptConsoleMessageLevel.WarningMessageLevel: + logger.warning(f"JS Console: {message}") + logger.debug(f" at {source_id}:{line_number}") + else: # ErrorMessageLevel + logger.error(f"JS Console: {message}") + logger.debug(f" at {source_id}:{line_number}") + + def _on_page_loaded(self, success: bool) -> None: + """Called when a page finishes loading. + + Checks if the bridge script was successfully injected. + + Args: + success: True if page loaded successfully + """ + if not success: + logger.warning("Page failed to load") + return + + # Check if bridge script is loaded + def check_script(result): + if result: + logger.info("โœ“ WebDrop Bridge script is active") + logger.info("โœ“ QWebChannel bridge is ready") + else: + logger.error("โœ— WebDrop Bridge script NOT loaded!") + logger.error(" Drag-and-drop conversion will NOT work") + + # Execute JS to check if our script is loaded + self.web_view.page().runJavaScript( + "typeof window.__webdrop_bridge_injected !== 'undefined' && window.__webdrop_bridge_injected === true", + check_script + ) def _create_navigation_toolbar(self) -> None: """Create navigation toolbar with Home, Back, Forward, Refresh buttons. @@ -488,7 +798,7 @@ class MainWindow(QMainWindow): Args: status: Status text to display - emoji: Optional emoji prefix (๐Ÿ”„, โœ…, โฌ‡๏ธ, โš ๏ธ) + emoji: Optional emoji prefix (rotating, checkmark, download, warning symbols) """ if emoji: self.update_status_label.setText(f"{emoji} {status}") @@ -559,24 +869,11 @@ class MainWindow(QMainWindow): # Can be extended with save operations or cleanup event.accept() - def initiate_drag(self, file_paths: list) -> bool: - """Initiate a drag operation for the given files. - - Called from web content via JavaScript bridge. - - Args: - file_paths: List of file paths to drag - - Returns: - True if drag was initiated successfully - """ - return self.drag_interceptor.initiate_drag(file_paths) - def check_for_updates_startup(self) -> None: """Check for updates on application startup. Runs asynchronously in background without blocking UI. - Uses 24h cache so won't hammer the API. + Uses 24-hour cache so will not hammer the API. """ from webdrop_bridge.core.updater import UpdateManager diff --git a/src/webdrop_bridge/ui/restricted_web_view.py b/src/webdrop_bridge/ui/restricted_web_view.py index d7b28cc..489b473 100644 --- a/src/webdrop_bridge/ui/restricted_web_view.py +++ b/src/webdrop_bridge/ui/restricted_web_view.py @@ -1,13 +1,106 @@ """Restricted web view with URL whitelist enforcement for Kiosk-mode.""" import fnmatch -from typing import List, Optional +import logging +from pathlib import Path +from typing import List, Optional, Union -from PySide6.QtCore import QUrl +from PySide6.QtCore import QStandardPaths, QUrl from PySide6.QtGui import QDesktopServices -from PySide6.QtWebEngineCore import QWebEngineNavigationRequest +from PySide6.QtWebEngineCore import QWebEngineNavigationRequest, QWebEnginePage, QWebEngineProfile from PySide6.QtWebEngineWidgets import QWebEngineView +logger = logging.getLogger(__name__) + + +logger = logging.getLogger(__name__) + + +class CustomWebEnginePage(QWebEnginePage): + """Custom page that handles new window requests for downloads.""" + + def acceptNavigationRequest( + self, url: Union[QUrl, str], nav_type: QWebEnginePage.NavigationType, is_main_frame: bool + ) -> bool: + """Handle navigation requests, including download links. + + Args: + url: Target URL (QUrl or string) + nav_type: Type of navigation (link click, form submit, etc.) + is_main_frame: Whether this is the main frame + + Returns: + True to accept navigation, False to reject + """ + # Convert to string if QUrl + url_str = url.toString() if isinstance(url, QUrl) else url + + # Log all navigation attempts for debugging + logger.debug(f"Navigation request: {url_str} (type={nav_type}, main_frame={is_main_frame})") + + # Check if this might be a download (common file extensions) + download_extensions = [ + ".pdf", + ".zip", + ".rar", + ".7z", + ".tar", + ".gz", + ".doc", + ".docx", + ".xls", + ".xlsx", + ".ppt", + ".pptx", + ".jpg", + ".jpeg", + ".png", + ".gif", + ".bmp", + ".svg", + ".mp4", + ".mp3", + ".avi", + ".mov", + ".wav", + ".exe", + ".msi", + ".dmg", + ".pkg", + ".csv", + ".txt", + ".json", + ".xml", + ] + + if any(url_str.lower().endswith(ext) for ext in download_extensions): + logger.info(f"๐Ÿ”ฝ Detected potential download URL: {url_str}") + # This will trigger downloadRequested if it's a download + + return super().acceptNavigationRequest(url, nav_type, is_main_frame) + + def createWindow(self, window_type: QWebEnginePage.WebWindowType) -> QWebEnginePage: + """Handle new window requests (target=_blank, window.open, etc.). + + Many downloads are triggered via target="_blank" links. + + Args: + window_type: Type of window being created + + Returns: + New page instance for the window + """ + logger.info(f"๐ŸชŸ New window requested, type: {window_type}") + + # Create a temporary page to handle the download + # This page will never be displayed but allows downloads to work + download_page = QWebEnginePage(self.profile(), self) + + logger.info("โœ… Created temporary page for download/popup") + + # Return the temporary page - it will trigger downloadRequested if it's a download + return download_page + class RestrictedWebEngineView(QWebEngineView): """Web view that enforces URL whitelist for Kiosk-mode security. @@ -27,31 +120,81 @@ class RestrictedWebEngineView(QWebEngineView): super().__init__() self.allowed_urls = allowed_urls or [] + # Create persistent profile for cookie and session storage + self.profile = self._create_persistent_profile() + + # Use custom page for better download handling with persistent profile + custom_page = CustomWebEnginePage(self.profile, self) + self.setPage(custom_page) + + logger.info( + "RestrictedWebEngineView initialized with CustomWebEnginePage and persistent profile" + ) + # Connect to navigation request handler self.page().navigationRequested.connect(self._on_navigation_requested) - def _on_navigation_requested( - self, request: QWebEngineNavigationRequest - ) -> None: + def _create_persistent_profile(self) -> QWebEngineProfile: + """Create and configure a persistent web engine profile. + + This enables persistent cookies and cache storage, allowing + authentication sessions (e.g., Microsoft login) to persist + across application restarts. + + Returns: + Configured QWebEngineProfile with persistent storage + """ + # Get application data directory + app_data_dir = QStandardPaths.writableLocation( + QStandardPaths.StandardLocation.AppDataLocation + ) + + # Create profile directory path + profile_path = Path(app_data_dir) / "WebEngineProfile" + profile_path.mkdir(parents=True, exist_ok=True) + + # Create persistent profile with custom storage location + # Using "WebDropBridge" as the profile name + profile = QWebEngineProfile("WebDropBridge", self) + profile.setPersistentStoragePath(str(profile_path)) + + # Configure persistent cookies (critical for authentication) + profile.setPersistentCookiesPolicy( + QWebEngineProfile.PersistentCookiesPolicy.ForcePersistentCookies + ) + + # Enable HTTP cache for better performance + profile.setHttpCacheType(QWebEngineProfile.HttpCacheType.DiskHttpCache) + + # Set cache size to 100 MB + profile.setHttpCacheMaximumSize(100 * 1024 * 1024) + + logger.info(f"Created persistent profile at: {profile_path}") + logger.info("Cookies policy: ForcePersistentCookies") + logger.info("HTTP cache: DiskHttpCache (100 MB)") + + return profile + + def _on_navigation_requested(self, request: QWebEngineNavigationRequest) -> None: """Handle navigation requests and enforce URL whitelist. Args: request: Navigation request to process """ - url = request.url + url = request.url() # If no restrictions, allow all URLs if not self.allowed_urls: return # Check if URL matches whitelist - if self._is_url_allowed(url): # type: ignore[operator] + if self._is_url_allowed(url): # Allow the navigation (default behavior) return # URL not whitelisted - open in system browser request.reject() - QDesktopServices.openUrl(url) # type: ignore[operator] + QDesktopServices.openUrl(url) def _is_url_allowed(self, url: QUrl) -> bool: """Check if a URL matches the whitelist. @@ -98,4 +241,3 @@ class RestrictedWebEngineView(QWebEngineView): return True return False - diff --git a/test_download.html b/test_download.html new file mode 100644 index 0000000..3b6719c --- /dev/null +++ b/test_download.html @@ -0,0 +1,95 @@ + + + + + + Download Test + + + +

๐Ÿงช Download Test Seite

+ +
+

Test 1: Direct Download Link

+

Download einer Text-Datei via direktem Link:

+ ๐Ÿ“ฅ Download test.txt +
+ +
+

Test 2: JavaScript Download (Blob)

+

Download via JavaScript Blob:

+ +
+ +
+

Test 3: Base64 Image Download

+

Download eines kleinen Bildes:

+ ๐Ÿ“ฅ Download test.png +
+ +
+

Test 4: External Link (should open in browser)

+

PDF von externem Server:

+ ๐Ÿ“ฅ Download external PDF +
+ + + + diff --git a/test_download.py b/test_download.py new file mode 100644 index 0000000..5b72aa4 --- /dev/null +++ b/test_download.py @@ -0,0 +1,19 @@ +"""Test script to verify download functionality.""" + +import sys +from pathlib import Path + +from PySide6.QtCore import QStandardPaths, QUrl +from PySide6.QtWebEngineCore import QWebEngineDownloadRequest +from PySide6.QtWidgets import QApplication + +# Test download path resolution +downloads_path = QStandardPaths.writableLocation(QStandardPaths.StandardLocation.DownloadLocation) +print(f"Downloads folder: {downloads_path}") +print(f"Downloads exists: {Path(downloads_path).exists()}") + +# Test file path construction +test_filename = "test_file.pdf" +download_file = Path(downloads_path) / test_filename +print(f"Test download path: {download_file}") +print(f"Parent exists: {download_file.parent.exists()}") diff --git a/test_url_mappings.py b/test_url_mappings.py new file mode 100644 index 0000000..0c6c3a7 --- /dev/null +++ b/test_url_mappings.py @@ -0,0 +1,24 @@ +"""Quick test to verify URL mappings are loaded correctly.""" +import sys +from pathlib import Path + +# Add src to path +sys.path.insert(0, str(Path(__file__).parent / "src")) + +from webdrop_bridge.config import Config + +# Load config from .env +config = Config.from_env() + +print(f"โœ“ Config loaded successfully") +print(f" URL Mappings: {len(config.url_mappings)} loaded") + +if config.url_mappings: + for i, mapping in enumerate(config.url_mappings, 1): + print(f" {i}. {mapping.url_prefix}") + print(f" -> {mapping.local_path}") +else: + print(" โš  No URL mappings found!") + +print(f"\n Allowed roots: {config.allowed_roots}") +print(f" Web app URL: {config.webapp_url}") diff --git a/tests/unit/test_config.py b/tests/unit/test_config.py index c8f569f..d20de67 100644 --- a/tests/unit/test_config.py +++ b/tests/unit/test_config.py @@ -98,12 +98,13 @@ class TestConfigFromEnv: Config.from_env(str(env_file)) def test_from_env_invalid_root_path(self, tmp_path): - """Test that non-existent root paths raise ConfigurationError.""" + """Test that non-existent root paths are logged as warning but don't raise error.""" env_file = tmp_path / ".env" env_file.write_text("ALLOWED_ROOTS=/nonexistent/path/that/does/not/exist\n") - with pytest.raises(ConfigurationError, match="does not exist"): - Config.from_env(str(env_file)) + # Should not raise - just logs warning and returns empty allowed_roots + config = Config.from_env(str(env_file)) + assert config.allowed_roots == [] # Non-existent roots are skipped def test_from_env_empty_webapp_url(self, tmp_path): """Test that empty webapp URL raises ConfigurationError.""" diff --git a/tests/unit/test_drag_interceptor.py b/tests/unit/test_drag_interceptor.py index d94fb23..eaa5ce4 100644 --- a/tests/unit/test_drag_interceptor.py +++ b/tests/unit/test_drag_interceptor.py @@ -3,63 +3,79 @@ from pathlib import Path from unittest.mock import MagicMock, patch +import pytest + +from webdrop_bridge.config import Config from webdrop_bridge.core.drag_interceptor import DragInterceptor -from webdrop_bridge.core.validator import PathValidator + + +@pytest.fixture +def test_config(tmp_path): + """Create test configuration.""" + return Config( + app_name="Test App", + app_version="1.0.0", + log_level="INFO", + log_file=None, + allowed_roots=[tmp_path], + allowed_urls=[], + webapp_url="https://wps.agravity.io/", + url_mappings=[], + check_file_exists=True, + ) class TestDragInterceptorInitialization: """Test DragInterceptor initialization and setup.""" - def test_drag_interceptor_creation(self, qtbot): + def test_drag_interceptor_creation(self, qtbot, test_config): """Test DragInterceptor can be instantiated.""" - interceptor = DragInterceptor() + interceptor = DragInterceptor(test_config) assert interceptor is not None - assert interceptor._validator is None + assert interceptor._validator is not None + assert interceptor._url_converter is not None - def test_drag_interceptor_has_signals(self, qtbot): + def test_drag_interceptor_has_signals(self, qtbot, test_config): """Test DragInterceptor has required signals.""" - interceptor = DragInterceptor() + interceptor = DragInterceptor(test_config) assert hasattr(interceptor, "drag_started") assert hasattr(interceptor, "drag_failed") - def test_set_validator(self, qtbot, tmp_path): - """Test setting validator on drag interceptor.""" - interceptor = DragInterceptor() - validator = PathValidator([tmp_path]) - - interceptor.set_validator(validator) - - assert interceptor._validator is validator + def test_set_validator(self, qtbot, test_config): + """Test validator is set during construction.""" + interceptor = DragInterceptor(test_config) + assert interceptor._validator is not None class TestDragInterceptorValidation: """Test path validation in drag operations.""" - def test_initiate_drag_no_files(self, qtbot): - """Test initiating drag with no files fails.""" - interceptor = DragInterceptor() + def test_handle_drag_empty_text(self, qtbot, test_config): + """Test handling drag with empty text fails.""" + interceptor = DragInterceptor(test_config) with qtbot.waitSignal(interceptor.drag_failed): - result = interceptor.initiate_drag([]) + result = interceptor.handle_drag("") assert result is False - def test_initiate_drag_no_validator(self, qtbot): - """Test initiating drag without validator fails.""" - interceptor = DragInterceptor() - with qtbot.waitSignal(interceptor.drag_failed): - result = interceptor.initiate_drag(["file.txt"]) - - assert result is False - - def test_initiate_drag_single_valid_file(self, qtbot, tmp_path): - """Test initiating drag with single valid file.""" + def test_handle_drag_valid_file_path(self, qtbot, tmp_path): + """Test handling drag with valid file path.""" # Create a test file test_file = tmp_path / "test.txt" test_file.write_text("test content") - interceptor = DragInterceptor() - validator = PathValidator([tmp_path]) - interceptor.set_validator(validator) + config = Config( + app_name="Test", + app_version="1.0.0", + log_level="INFO", + log_file=None, + allowed_roots=[tmp_path], + allowed_urls=[], + webapp_url="https://test.com/", + url_mappings=[], + check_file_exists=True, + ) + interceptor = DragInterceptor(config) # Mock the drag operation to simulate success with patch("webdrop_bridge.core.drag_interceptor.QDrag") as mock_drag: @@ -69,114 +85,91 @@ class TestDragInterceptorValidation: mock_drag_instance.exec.return_value = Qt.DropAction.CopyAction mock_drag.return_value = mock_drag_instance - result = interceptor.initiate_drag([str(test_file)]) + result = interceptor.handle_drag(str(test_file)) # Should return True on successful drag assert result is True - def test_initiate_drag_invalid_path(self, qtbot, tmp_path): + def test_handle_drag_invalid_path(self, qtbot, test_config): """Test drag with invalid path fails.""" - interceptor = DragInterceptor() - validator = PathValidator([tmp_path]) - interceptor.set_validator(validator) + interceptor = DragInterceptor(test_config) # Path outside allowed roots - invalid_path = Path("/etc/passwd") + invalid_path = "/etc/passwd" with qtbot.waitSignal(interceptor.drag_failed): - result = interceptor.initiate_drag([str(invalid_path)]) + result = interceptor.handle_drag(invalid_path) assert result is False - def test_initiate_drag_nonexistent_file(self, qtbot, tmp_path): + def test_handle_drag_nonexistent_file(self, qtbot, test_config, tmp_path): """Test drag with nonexistent file fails.""" - interceptor = DragInterceptor() - validator = PathValidator([tmp_path]) - interceptor.set_validator(validator) + interceptor = DragInterceptor(test_config) nonexistent = tmp_path / "nonexistent.txt" with qtbot.waitSignal(interceptor.drag_failed): - result = interceptor.initiate_drag([str(nonexistent)]) + result = interceptor.handle_drag(str(nonexistent)) assert result is False -class TestDragInterceptorMultipleFiles: - """Test drag operations with multiple files.""" +class TestDragInterceptorAzureURL: + """Test Azure URL to local path conversion in drag operations.""" - def test_initiate_drag_multiple_files(self, qtbot, tmp_path): - """Test drag with multiple valid files.""" - # Create test files - file1 = tmp_path / "file1.txt" - file2 = tmp_path / "file2.txt" - file1.write_text("content 1") - file2.write_text("content 2") + def test_handle_drag_azure_url(self, qtbot, tmp_path): + """Test handling drag with Azure Blob Storage URL.""" + from webdrop_bridge.config import URLMapping - interceptor = DragInterceptor() - validator = PathValidator([tmp_path]) - interceptor.set_validator(validator) + # Create test file that would be the result + test_file = tmp_path / "test.png" + test_file.write_text("image data") - from PySide6.QtCore import Qt + config = Config( + app_name="Test", + app_version="1.0.0", + log_level="INFO", + log_file=None, + allowed_roots=[tmp_path], + allowed_urls=[], + webapp_url="https://test.com/", + url_mappings=[ + URLMapping( + url_prefix="https://wpsagravitystg.file.core.windows.net/wpsagravitysync/", + local_path=str(tmp_path) + ) + ], + check_file_exists=True, + ) + interceptor = DragInterceptor(config) + + # Azure URL + azure_url = "https://wpsagravitystg.file.core.windows.net/wpsagravitysync/test.png" + + # Mock the drag operation with patch("webdrop_bridge.core.drag_interceptor.QDrag") as mock_drag: mock_drag_instance = MagicMock() + from PySide6.QtCore import Qt mock_drag_instance.exec.return_value = Qt.DropAction.CopyAction mock_drag.return_value = mock_drag_instance - result = interceptor.initiate_drag([str(file1), str(file2)]) + result = interceptor.handle_drag(azure_url) assert result is True - def test_initiate_drag_mixed_valid_invalid(self, qtbot, tmp_path): - """Test drag with mix of valid and invalid paths fails.""" - test_file = tmp_path / "valid.txt" - test_file.write_text("content") + def test_handle_drag_unmapped_url(self, qtbot, test_config): + """Test handling drag with unmapped URL fails.""" + interceptor = DragInterceptor(test_config) - interceptor = DragInterceptor() - validator = PathValidator([tmp_path]) - interceptor.set_validator(validator) + # URL with no mapping + unmapped_url = "https://unknown.blob.core.windows.net/container/file.png" - # Mix of valid and invalid paths with qtbot.waitSignal(interceptor.drag_failed): - result = interceptor.initiate_drag( - [str(test_file), "/etc/passwd"] - ) + result = interceptor.handle_drag(unmapped_url) assert result is False -class TestDragInterceptorMimeData: - """Test MIME data creation and file URL formatting.""" - - def test_mime_data_creation(self, qtbot, tmp_path): - """Test MIME data is created with proper file URLs.""" - test_file = tmp_path / "test.txt" - test_file.write_text("test") - - interceptor = DragInterceptor() - validator = PathValidator([tmp_path]) - interceptor.set_validator(validator) - - from PySide6.QtCore import Qt - with patch("webdrop_bridge.core.drag_interceptor.QDrag") as mock_drag: - mock_drag_instance = MagicMock() - mock_drag_instance.exec.return_value = Qt.DropAction.CopyAction - mock_drag.return_value = mock_drag_instance - - interceptor.initiate_drag([str(test_file)]) - - # Check MIME data was set correctly - call_args = mock_drag_instance.setMimeData.call_args - mime_data = call_args[0][0] - - # Verify URLs were set - urls = mime_data.urls() - assert len(urls) == 1 - # Check that the URL contains file:// scheme (can be string repr or QUrl) - url_str = str(urls[0]).lower() - assert "file://" in url_str - - class TestDragInterceptorSignals: """Test signal emission on drag operations.""" @@ -185,153 +178,60 @@ class TestDragInterceptorSignals: test_file = tmp_path / "test.txt" test_file.write_text("content") - interceptor = DragInterceptor() - validator = PathValidator([tmp_path]) - interceptor.set_validator(validator) - - from PySide6.QtCore import Qt + config = Config( + app_name="Test", + app_version="1.0.0", + log_level="INFO", + log_file=None, + allowed_roots=[tmp_path], + allowed_urls=[], + webapp_url="https://test.com/", + url_mappings=[], + check_file_exists=True, + ) + interceptor = DragInterceptor(config) # Connect to signal manually signal_spy = [] - interceptor.drag_started.connect(lambda paths: signal_spy.append(paths)) + interceptor.drag_started.connect(lambda src, path: signal_spy.append((src, path))) + from PySide6.QtCore import Qt with patch("webdrop_bridge.core.drag_interceptor.QDrag") as mock_drag: mock_drag_instance = MagicMock() mock_drag_instance.exec.return_value = Qt.DropAction.CopyAction mock_drag.return_value = mock_drag_instance - result = interceptor.initiate_drag([str(test_file)]) + result = interceptor.handle_drag(str(test_file)) # Verify result and signal emission assert result is True assert len(signal_spy) == 1 - def test_drag_failed_signal_on_no_files(self, qtbot): - """Test drag_failed signal on empty file list.""" - interceptor = DragInterceptor() + def test_drag_failed_signal_on_empty_text(self, qtbot, test_config): + """Test drag_failed signal on empty text.""" + interceptor = DragInterceptor(test_config) # Connect to signal manually signal_spy = [] - interceptor.drag_failed.connect(lambda msg: signal_spy.append(msg)) + interceptor.drag_failed.connect(lambda src, msg: signal_spy.append((src, msg))) - result = interceptor.initiate_drag([]) + result = interceptor.handle_drag("") # Verify result and signal emission assert result is False assert len(signal_spy) == 1 - assert "No files" in signal_spy[0] + assert "Empty" in signal_spy[0][1] - def test_drag_failed_signal_on_validation_error(self, qtbot, tmp_path): + def test_drag_failed_signal_on_validation_error(self, qtbot, test_config): """Test drag_failed signal on validation failure.""" - interceptor = DragInterceptor() - validator = PathValidator([tmp_path]) - interceptor.set_validator(validator) + interceptor = DragInterceptor(test_config) # Connect to signal manually signal_spy = [] - interceptor.drag_failed.connect(lambda msg: signal_spy.append(msg)) + interceptor.drag_failed.connect(lambda src, msg: signal_spy.append((src, msg))) - result = interceptor.initiate_drag(["/invalid/path/file.txt"]) + result = interceptor.handle_drag("/invalid/path/file.txt") # Verify result and signal emission assert result is False assert len(signal_spy) == 1 - - -class TestDragInterceptorDragExecution: - """Test drag operation execution and result handling.""" - - def test_drag_cancelled_returns_false(self, qtbot, tmp_path): - """Test drag cancellation returns False.""" - test_file = tmp_path / "test.txt" - test_file.write_text("content") - - interceptor = DragInterceptor() - validator = PathValidator([tmp_path]) - interceptor.set_validator(validator) - - with patch("webdrop_bridge.core.drag_interceptor.QDrag") as mock_drag: - mock_drag_instance = MagicMock() - mock_drag_instance.exec.return_value = 0 # Cancelled/failed - mock_drag.return_value = mock_drag_instance - - # Connect to signal manually - signal_spy = [] - interceptor.drag_failed.connect(lambda msg: signal_spy.append(msg)) - - result = interceptor.initiate_drag([str(test_file)]) - - assert result is False - assert len(signal_spy) == 1 - - def test_pixmap_created_from_widget(self, qtbot, tmp_path): - """Test pixmap is created from widget grab.""" - test_file = tmp_path / "test.txt" - test_file.write_text("content") - - interceptor = DragInterceptor() - validator = PathValidator([tmp_path]) - interceptor.set_validator(validator) - - from PySide6.QtCore import Qt - with patch.object(interceptor, "grab") as mock_grab: - mock_pixmap = MagicMock() - mock_grab.return_value.scaled.return_value = mock_pixmap - - with patch("webdrop_bridge.core.drag_interceptor.QDrag") as mock_drag: - mock_drag_instance = MagicMock() - mock_drag_instance.exec.return_value = Qt.DropAction.CopyAction - mock_drag.return_value = mock_drag_instance - - interceptor.initiate_drag([str(test_file)]) - - # Verify grab was called and pixmap was set - mock_grab.assert_called_once() - mock_drag_instance.setPixmap.assert_called_once_with(mock_pixmap) - - -class TestDragInterceptorIntegration: - """Integration tests with PathValidator.""" - - def test_drag_with_nested_file(self, qtbot, tmp_path): - """Test drag with file in nested directory.""" - nested_dir = tmp_path / "nested" / "dir" - nested_dir.mkdir(parents=True) - test_file = nested_dir / "file.txt" - test_file.write_text("nested content") - - interceptor = DragInterceptor() - validator = PathValidator([tmp_path]) - interceptor.set_validator(validator) - - from PySide6.QtCore import Qt - with patch("webdrop_bridge.core.drag_interceptor.QDrag") as mock_drag: - mock_drag_instance = MagicMock() - mock_drag_instance.exec.return_value = Qt.DropAction.CopyAction - mock_drag.return_value = mock_drag_instance - - result = interceptor.initiate_drag([str(test_file)]) - - assert result is True - - def test_drag_with_relative_path(self, qtbot, tmp_path): - """Test drag with relative path resolution.""" - test_file = tmp_path / "relative.txt" - test_file.write_text("content") - - interceptor = DragInterceptor() - validator = PathValidator([tmp_path]) - interceptor.set_validator(validator) - - # This would work if run from the directory, but we'll just verify - # the interceptor handles Path objects correctly - from PySide6.QtCore import Qt - with patch("webdrop_bridge.core.drag_interceptor.QDrag") as mock_drag: - mock_drag_instance = MagicMock() - mock_drag_instance.exec.return_value = Qt.DropAction.CopyAction - mock_drag.return_value = mock_drag_instance - - # Direct absolute path for reliable test - result = interceptor.initiate_drag([str(test_file)]) - - assert result is True diff --git a/tests/unit/test_main_window.py b/tests/unit/test_main_window.py index 75216e0..72a53d3 100644 --- a/tests/unit/test_main_window.py +++ b/tests/unit/test_main_window.py @@ -231,10 +231,12 @@ class TestMainWindowDragIntegration: assert window.drag_interceptor.drag_started is not None assert window.drag_interceptor.drag_failed is not None - def test_initiate_drag_delegates_to_interceptor( + def test_handle_drag_delegates_to_interceptor( self, qtbot, sample_config, tmp_path ): - """Test initiate_drag method delegates to interceptor.""" + """Test drag handling delegates to interceptor.""" + from PySide6.QtCore import QCoreApplication + window = MainWindow(sample_config) qtbot.addWidget(window) @@ -243,29 +245,32 @@ class TestMainWindowDragIntegration: test_file.write_text("test") with patch.object( - window.drag_interceptor, "initiate_drag" + window.drag_interceptor, "handle_drag" ) as mock_drag: mock_drag.return_value = True - result = window.initiate_drag([str(test_file)]) + # Call through bridge + window._drag_bridge.start_file_drag(str(test_file)) + + # Process deferred QTimer.singleShot(0, ...) call + QCoreApplication.processEvents() - mock_drag.assert_called_once_with([str(test_file)]) - assert result is True + mock_drag.assert_called_once_with(str(test_file)) def test_on_drag_started_called(self, qtbot, sample_config): """Test _on_drag_started handler can be called.""" window = MainWindow(sample_config) qtbot.addWidget(window) - # Should not raise - window._on_drag_started(["/some/path"]) + # Should not raise - new signature has source and local_path + window._on_drag_started("https://example.com/file.png", "/local/path/file.png") def test_on_drag_failed_called(self, qtbot, sample_config): """Test _on_drag_failed handler can be called.""" window = MainWindow(sample_config) qtbot.addWidget(window) - # Should not raise - window._on_drag_failed("Test error message") + # Should not raise - new signature has source and error + window._on_drag_failed("https://example.com/file.png", "Test error message") class TestMainWindowURLWhitelist: diff --git a/tests/unit/test_url_converter.py b/tests/unit/test_url_converter.py new file mode 100644 index 0000000..60ab58f --- /dev/null +++ b/tests/unit/test_url_converter.py @@ -0,0 +1,144 @@ +"""Unit tests for URL converter.""" + +from pathlib import Path + +import pytest + +from webdrop_bridge.config import Config, URLMapping +from webdrop_bridge.core.url_converter import URLConverter + + +@pytest.fixture +def test_config(): + """Create test configuration with URL mappings.""" + return Config( + app_name="Test App", + app_version="1.0.0", + log_level="INFO", + log_file=None, + allowed_roots=[], + allowed_urls=[], + webapp_url="https://wps.agravity.io/", + url_mappings=[ + URLMapping( + url_prefix="https://wpsagravitystg.file.core.windows.net/wpsagravitysync/", + local_path="Z:" + ), + URLMapping( + url_prefix="https://other.blob.core.windows.net/container/", + local_path="Y:\\shared" + ) + ] + ) + + +@pytest.fixture +def converter(test_config): + """Create URL converter with test config.""" + return URLConverter(test_config) + + +def test_convert_simple_url(converter): + """Test converting a simple Azure URL to local path.""" + url = "https://wpsagravitystg.file.core.windows.net/wpsagravitysync/test/file.png" + result = converter.convert_url_to_path(url) + + assert result is not None + assert str(result).endswith("test\\file.png") # Windows path separator + + +def test_convert_url_with_special_characters(converter): + """Test URL with special characters (URL encoded).""" + url = "https://wpsagravitystg.file.core.windows.net/wpsagravitysync/folder/file%20with%20spaces.png" + result = converter.convert_url_to_path(url) + + assert result is not None + assert "file with spaces.png" in str(result) + + +def test_convert_url_with_subdirectories(converter): + """Test URL with deep directory structure.""" + url = "https://wpsagravitystg.file.core.windows.net/wpsagravitysync/aN5PysnXIuRECzcRbvHkjL7g0/subfolder/file.png" + result = converter.convert_url_to_path(url) + + assert result is not None + assert "aN5PysnXIuRECzcRbvHkjL7g0" in str(result) + assert "subfolder" in str(result) + + +def test_convert_unmapped_url(converter): + """Test URL that doesn't match any mapping.""" + url = "https://unknown.blob.core.windows.net/container/file.png" + result = converter.convert_url_to_path(url) + + assert result is None + + +def test_convert_empty_url(converter): + """Test empty URL.""" + result = converter.convert_url_to_path("") + assert result is None + + +def test_convert_none_url(converter): + """Test None URL.""" + result = converter.convert_url_to_path(None) + assert result is None + + +def test_is_azure_url_positive(converter): + """Test recognizing valid Azure URLs.""" + url = "https://wpsagravitystg.file.core.windows.net/wpsagravitysync/file.png" + assert converter.is_azure_url(url) is True + + +def test_is_azure_url_negative(converter): + """Test rejecting non-Azure URLs.""" + assert converter.is_azure_url("https://example.com/file.png") is False + assert converter.is_azure_url("Z:\\file.png") is False + assert converter.is_azure_url("") is False + + +def test_multiple_mappings(converter): + """Test that correct mapping is used for URL.""" + url1 = "https://wpsagravitystg.file.core.windows.net/wpsagravitysync/file.png" + url2 = "https://other.blob.core.windows.net/container/file.png" + + result1 = converter.convert_url_to_path(url1) + result2 = converter.convert_url_to_path(url2) + + assert result1 is not None + assert result2 is not None + assert str(result1).startswith("Z:") + assert str(result2).startswith("Y:") + + +def test_url_mapping_validation_http(): + """Test that URL mapping requires http:// or https://.""" + with pytest.raises(Exception): # ConfigurationError + URLMapping( + url_prefix="ftp://server/path/", + local_path="Z:" + ) + + +def test_url_mapping_adds_trailing_slash(): + """Test that URL mapping adds trailing slash if missing.""" + mapping = URLMapping( + url_prefix="https://example.com/path", + local_path="Z:" + ) + assert mapping.url_prefix.endswith("/") + + +def test_convert_url_example_from_docs(converter): + """Test the exact example from documentation.""" + url = "https://wpsagravitystg.file.core.windows.net/wpsagravitysync/aN5PysnXIuRECzcRbvHkjL7g0/Hintergrund_Agravity.png" + result = converter.convert_url_to_path(url) + + assert result is not None + # Should be: Z:\aN5PysnXIuRECzcRbvHkjL7g0\Hintergrund_Agravity.png + expected_parts = ["Z:", "aN5PysnXIuRECzcRbvHkjL7g0", "Hintergrund_Agravity.png"] + result_str = str(result) + for part in expected_parts: + assert part in result_str diff --git a/tests/unit/test_validator.py b/tests/unit/test_validator.py index db8ae61..eabb228 100644 --- a/tests/unit/test_validator.py +++ b/tests/unit/test_validator.py @@ -22,11 +22,12 @@ class TestPathValidator: assert len(validator.allowed_roots) == 2 def test_validator_nonexistent_root(self, tmp_path): - """Test that nonexistent root raises ValidationError.""" + """Test that nonexistent root is logged as warning but doesn't raise error.""" nonexistent = tmp_path / "nonexistent" - with pytest.raises(ValidationError, match="does not exist"): - PathValidator([nonexistent]) + # Should not raise - just logs warning and skips the root + validator = PathValidator([nonexistent]) + assert len(validator.allowed_roots) == 0 # Non-existent roots are skipped def test_validator_non_directory_root(self, tmp_path): """Test that non-directory root raises ValidationError.""" diff --git a/webapp/index.html b/webapp/index.html index ac302bf..b675dc3 100644 --- a/webapp/index.html +++ b/webapp/index.html @@ -162,20 +162,26 @@
๐Ÿ–ผ๏ธ
-

Sample Image

+

Local Z:\ Image

Z:\data\test-image.jpg

๐Ÿ“„
-

Sample Document

+

Local Z:\ Document

Z:\data\API_DOCUMENTATION.pdf

-
๐Ÿ“Š
-

Sample Data

-

C:\Users\Public\data.csv

+
โ˜๏ธ
+

Azure Blob Storage Image

+

https://wpsagravitystg.file.core.windows.net/wpsagravitysync/aN5PysnXIuRECzcRbvHkjL7g0/Hintergrund_Agravity.png

+
+ +
+
โ˜๏ธ
+

Azure Blob Storage Document

+

https://wpsagravitystg.file.core.windows.net/wpsagravitysync/test/document.pdf

@@ -183,15 +189,59 @@

How to test:

  1. Open InDesign, Word, or Notepad++
    2. -
  2. Drag one of the items below to the application
    3. -
  3. The file path should be converted to a real file drag
    4. +
  4. Drag one of the items above to the application
    5. +
  5. Local Z:\ paths and Azure URLs will be converted to file drags
    6. +
  6. Azure URLs will be mapped to Z:\ paths automatically
  7. Check the browser console (F12) for debug info
+

Note: When dragging images from web apps like Agravity, the browser may not provide text/plain data. Press ALT while dragging to force text drag mode.

WebDrop Bridge v1.0.0 | Built with Qt and PySide6

+ +