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.
This commit is contained in:
claudi 2026-02-17 15:56:53 +01:00
parent c9704efc8d
commit 88dc358894
21 changed files with 1870 additions and 432 deletions

209
CONFIG_README.md Normal file
View file

@ -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"
}
```

22
config.example.json Normal file
View file

@ -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
}

15
config_test.json Normal file
View file

@ -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
}

View file

@ -1,8 +1,9 @@
"""Configuration management for WebDrop Bridge application.""" """Configuration management for WebDrop Bridge application."""
import json
import logging import logging
import os import os
from dataclasses import dataclass from dataclasses import dataclass, field
from pathlib import Path from pathlib import Path
from typing import List from typing import List
@ -17,9 +18,29 @@ class ConfigurationError(Exception):
pass 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 @dataclass
class Config: class Config:
"""Application configuration loaded from environment variables. """Application configuration loaded from environment variables or JSON file.
Attributes: Attributes:
app_name: Application display name app_name: Application display name
@ -28,7 +49,11 @@ class Config:
log_file: Optional log file path log_file: Optional log file path
allowed_roots: List of whitelisted root directories for file access allowed_roots: List of whitelisted root directories for file access
allowed_urls: List of whitelisted URL domains/patterns (empty = no restriction) 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_width: Initial window width in pixels
window_height: Initial window height in pixels window_height: Initial window height in pixels
window_title: Main window title (default: "{app_name} v{app_version}") window_title: Main window title (default: "{app_name} v{app_version}")
@ -45,10 +70,85 @@ class Config:
allowed_roots: List[Path] allowed_roots: List[Path]
allowed_urls: List[str] allowed_urls: List[str]
webapp_url: str webapp_url: str
window_width: int url_mappings: List[URLMapping] = field(default_factory=list)
window_height: int check_file_exists: bool = True
window_title: str auto_check_updates: bool = True
enable_logging: bool 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 @classmethod
def from_env(cls, env_file: str | None = None) -> "Config": 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") log_file_str = os.getenv("LOG_FILE", "logs/webdrop_bridge.log")
allowed_roots_str = os.getenv("ALLOWED_ROOTS", "Z:/,C:/Users/Public") allowed_roots_str = os.getenv("ALLOWED_ROOTS", "Z:/,C:/Users/Public")
allowed_urls_str = os.getenv("ALLOWED_URLS", "") 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_width = int(os.getenv("WINDOW_WIDTH", "1024"))
window_height = int(os.getenv("WINDOW_HEIGHT", "768")) window_height = int(os.getenv("WINDOW_HEIGHT", "768"))
# Window title defaults to app_name + version if not specified # Window title defaults to app_name + version if not specified
@ -103,14 +203,13 @@ class Config:
for p in allowed_roots_str.split(","): for p in allowed_roots_str.split(","):
root_path = Path(p.strip()).resolve() root_path = Path(p.strip()).resolve()
if not root_path.exists(): if not root_path.exists():
raise ConfigurationError( logger.warning(f"Allowed root does not exist: {p.strip()}")
f"Allowed root '{p.strip()}' does not exist" elif not root_path.is_dir():
)
if not root_path.is_dir():
raise ConfigurationError( raise ConfigurationError(
f"Allowed root '{p.strip()}' is not a directory" f"Allowed root '{p.strip()}' is not a directory"
) )
allowed_roots.append(root_path) else:
allowed_roots.append(root_path)
except ConfigurationError: except ConfigurationError:
raise raise
except (ValueError, OSError) as e: except (ValueError, OSError) as e:
@ -140,6 +239,32 @@ class Config:
if url.strip() if url.strip()
] if allowed_urls_str else [] ] 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( return cls(
app_name=app_name, app_name=app_name,
app_version=app_version, app_version=app_version,
@ -148,12 +273,60 @@ class Config:
allowed_roots=allowed_roots, allowed_roots=allowed_roots,
allowed_urls=allowed_urls, allowed_urls=allowed_urls,
webapp_url=webapp_url, webapp_url=webapp_url,
url_mappings=url_mappings,
window_width=window_width, window_width=window_width,
window_height=window_height, window_height=window_height,
window_title=window_title, window_title=window_title,
enable_logging=enable_logging, 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: def __repr__(self) -> str:
"""Return developer-friendly representation.""" """Return developer-friendly representation."""
return ( return (

View file

@ -1,5 +1,6 @@
"""Qt widget for intercepting drag events and initiating native drag operations.""" """Qt widget for intercepting drag events and initiating native drag operations."""
import logging
from pathlib import Path from pathlib import Path
from typing import List, Optional from typing import List, Optional
@ -7,98 +8,129 @@ from PySide6.QtCore import QMimeData, Qt, QUrl, Signal
from PySide6.QtGui import QDrag from PySide6.QtGui import QDrag
from PySide6.QtWidgets import QWidget 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 from webdrop_bridge.core.validator import PathValidator, ValidationError
logger = logging.getLogger(__name__)
class DragInterceptor(QWidget): 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 Intercepts drag events from web content, converts Azure Blob Storage URLs
operations, allowing files to be dragged from web content to native to local paths, validates them, and initiates native Qt drag operations.
applications.
Signals: Signals:
drag_started: Emitted when a drag operation begins successfully drag_started: Emitted when a drag operation begins successfully
drag_failed: Emitted when drag initiation fails drag_failed: Emitted when drag initiation fails
""" """
# Signals with string parameters (file paths that were dragged) # Signals with string parameters
drag_started = Signal(list) # List[str] - list of file paths drag_started = Signal(str, str) # (url_or_path, local_path)
drag_failed = Signal(str) # str - error message 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. """Initialize the drag interceptor.
Args: Args:
config: Application configuration
parent: Parent widget parent: Parent widget
""" """
super().__init__(parent) 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: def handle_drag(self, text: str) -> bool:
"""Set the path validator for this interceptor. """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: Args:
validator: PathValidator instance to use for validation text: Azure Blob Storage URL or file path from web drag
"""
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
Returns: Returns:
True if drag was successfully initiated, False otherwise True if native drag was initiated, False otherwise
""" """
if not file_paths: if not text or not text.strip():
self.drag_failed.emit("No files to drag") error_msg = "Empty drag text"
logger.warning(error_msg)
self.drag_failed.emit("", error_msg)
return False return False
if not self._validator: text = text.strip()
self.drag_failed.emit("Validator not configured") logger.debug(f"Handling drag for text: {text}")
return False
# Validate all paths first # Check if it's an Azure URL and convert to local path
validated_paths = [] if self._url_converter.is_azure_url(text):
for path_str in file_paths: local_path = self._url_converter.convert_url_to_path(text)
try: if local_path is None:
path = Path(path_str) error_msg = "No mapping found for URL"
if self._validator.validate(path): logger.warning(f"{error_msg}: {text}")
validated_paths.append(path) self.drag_failed.emit(text, error_msg)
except ValidationError as e:
self.drag_failed.emit(f"Validation failed for {path_str}: {e}")
return False return False
source_text = text
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
else: 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 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

View file

@ -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

View file

@ -1,7 +1,10 @@
"""Path validation for secure file operations.""" """Path validation for secure file operations."""
import logging
from pathlib import Path from pathlib import Path
from typing import List from typing import List, Optional
logger = logging.getLogger(__name__)
class ValidationError(Exception): class ValidationError(Exception):
@ -18,28 +21,27 @@ class PathValidator:
directory traversal attacks. 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. """Initialize validator with allowed root directories.
Args: Args:
allowed_roots: List of Path objects representing allowed root dirs allowed_roots: List of Path objects representing allowed root dirs
check_file_exists: Whether to validate that files exist
Raises: Raises:
ValidationError: If any root doesn't exist or isn't a directory ValidationError: If any root doesn't exist or isn't a directory
""" """
self.allowed_roots = [] self.allowed_roots = []
self.check_file_exists = check_file_exists
for root in allowed_roots: for root in allowed_roots:
root_path = Path(root).resolve() root_path = Path(root).resolve()
if not root_path.exists(): if not root_path.exists():
raise ValidationError( logger.warning(f"Allowed root '{root}' does not exist")
f"Allowed root '{root}' does not exist" elif not root_path.is_dir():
) raise ValidationError(f"Allowed root '{root}' is not a directory")
if not root_path.is_dir(): else:
raise ValidationError( self.allowed_roots.append(root_path)
f"Allowed root '{root}' is not a directory"
)
self.allowed_roots.append(root_path)
def validate(self, path: Path) -> bool: def validate(self, path: Path) -> bool:
"""Validate that path is within an allowed root directory. """Validate that path is within an allowed root directory.
@ -59,28 +61,32 @@ class PathValidator:
except (OSError, ValueError) as e: except (OSError, ValueError) as e:
raise ValidationError(f"Cannot resolve path '{path}': {e}") from e raise ValidationError(f"Cannot resolve path '{path}': {e}") from e
# Check file exists # Check file exists if required
if not file_path.exists(): if self.check_file_exists:
raise ValidationError(f"File does not exist: {path}") 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) # Check it's a regular file (not directory, symlink to dir, etc)
if not file_path.is_file(): if not file_path.is_file():
raise ValidationError(f"Path is not a regular file: {path}") raise ValidationError(f"Path is not a regular file: {path}")
# Check path is within an allowed root # Check path is within an allowed root (if roots configured)
for allowed_root in self.allowed_roots: if self.allowed_roots:
try: for allowed_root in self.allowed_roots:
# This raises ValueError if file_path is not relative to root try:
file_path.relative_to(allowed_root) # This raises ValueError if file_path is not relative to root
return True file_path.relative_to(allowed_root)
except ValueError: return True
continue except ValueError:
continue
# Not in any allowed root # Not in any allowed root
raise ValidationError( raise ValidationError(
f"Path '{file_path}' is not within allowed roots: " f"Path '{file_path}' is not within allowed roots: "
f"{self.allowed_roots}" f"{self.allowed_roots}"
) )
return True
def is_valid(self, path: Path) -> bool: def is_valid(self, path: Path) -> bool:
"""Check if path is valid without raising exception. """Check if path is valid without raising exception.

View file

@ -19,8 +19,12 @@ def main() -> int:
int: Exit code (0 for success, non-zero for error) int: Exit code (0 for success, non-zero for error)
""" """
try: try:
# Load configuration from environment # Load configuration from file if it exists, otherwise from environment
config = Config.from_env() 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 # Set up logging
log_file = None log_file = None

View file

@ -5,6 +5,58 @@
if (window.__webdrop_bridge_injected) return; if (window.__webdrop_bridge_injected) return;
window.__webdrop_bridge_injected = true; 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) { function ensureChannel(cb) {
if (window.bridge) { cb(); return; } if (window.bridge) { cb(); return; }
@ -12,62 +64,151 @@
if (window.QWebChannel && window.qt && window.qt.webChannelTransport) { if (window.QWebChannel && window.qt && window.qt.webChannelTransport) {
new QWebChannel(window.qt.webChannelTransport, function(channel) { new QWebChannel(window.qt.webChannelTransport, function(channel) {
window.bridge = channel.objects.bridge; window.bridge = channel.objects.bridge;
console.log('[WebDrop Bridge] QWebChannel connected');
cb(); 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) { if (window.QWebChannel) {
init(); 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() { function hook() {
document.addEventListener('dragstart', function(e) { console.log('[WebDrop Bridge] Installing hook, have ' + webAppDragHandlers.length + ' intercepted handlers');
var dt = e.dataTransfer;
if (!dt) return;
// Get path from existing payload or from the card markup. if (window.bridge && typeof window.bridge.debug_log === 'function') {
var path = dt.getData('text/plain'); window.bridge.debug_log('Installing drag interceptor with ' + webAppDragHandlers.length + ' intercepted handlers');
if (!path) { }
var card = e.target.closest && e.target.closest('.drag-item');
if (card) { // Stop intercepting addEventListener - from now on, listeners register normally
var pathEl = card.querySelector('p'); listenerPatchActive = false;
if (pathEl) {
path = (pathEl.textContent || '').trim(); // 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
// Ensure text payload exists for non-file drags and downstream targets. // Reset state on dragend
if (!dt.getData('text/plain')) { originalAddEventListener.call(document, 'dragend', function(e) {
dt.setData('text/plain', path); currentDragData = null;
} }, false);
// Check if path is Z:\ — if yes, trigger native file drag. Otherwise, stay as text. console.log('[WebDrop Bridge] Drag listener registered on document (capture phase)');
var isZDrive = /^z:/i.test(path); }
if (!isZDrive) return;
// 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();
// Z:\ detected — prevent default browser drag and convert to native file drag
e.preventDefault();
ensureChannel(function() { ensureChannel(function() {
if (window.bridge && typeof window.bridge.start_file_drag === 'function') { if (window.bridge && typeof window.bridge.debug_log === 'function') {
window.bridge.start_file_drag(path); 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') { if (document.readyState === 'loading') {
document.addEventListener('DOMContentLoaded', hook); console.log('[WebDrop Bridge] Waiting for DOMContentLoaded...');
originalAddEventListener.call(document, 'DOMContentLoaded', installHook);
} else { } else {
hook(); console.log('[WebDrop Bridge] DOM already ready, installing hook...');
installHook();
} }
})(); })();

View file

@ -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');
})();

View file

@ -6,10 +6,22 @@ from datetime import datetime
from pathlib import Path from pathlib import Path
from typing import Optional from typing import Optional
from PySide6.QtCore import QObject, QPoint, QSize, Qt, QThread, QTimer, QUrl, Signal, Slot from PySide6.QtCore import (
from PySide6.QtGui import QIcon QEvent,
QObject,
QPoint,
QSize,
QStandardPaths,
Qt,
QThread,
QTimer,
QUrl,
Signal,
Slot,
)
from PySide6.QtGui import QIcon, QMouseEvent
from PySide6.QtWebChannel import QWebChannel from PySide6.QtWebChannel import QWebChannel
from PySide6.QtWebEngineCore import QWebEngineScript from PySide6.QtWebEngineCore import QWebEngineDownloadRequest, QWebEngineScript
from PySide6.QtWidgets import ( from PySide6.QtWidgets import (
QLabel, QLabel,
QMainWindow, QMainWindow,
@ -202,19 +214,29 @@ class _DragBridge(QObject):
@Slot(str) @Slot(str)
def start_file_drag(self, path_text: str) -> None: 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. Defers execution to avoid Qt drag manager state issues.
Args: 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}") logger.debug(f"Bridge: start_file_drag called for {path_text}")
# Defer to avoid drag manager state issues # Defer to avoid drag manager state issues
# initiate_drag() handles validation internally # handle_drag() handles URL conversion and validation internally
QTimer.singleShot(0, lambda: self.window.drag_interceptor.initiate_drag([path_text])) 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): class MainWindow(QMainWindow):
@ -258,17 +280,24 @@ class MainWindow(QMainWindow):
# Create web engine view # Create web engine view
self.web_view = RestrictedWebEngineView(config.allowed_urls) 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) # Create navigation toolbar (Kiosk-mode navigation)
self._create_navigation_toolbar() self._create_navigation_toolbar()
# Create status bar # Create status bar
self._create_status_bar() self._create_status_bar()
# Create drag interceptor # Create drag interceptor with config (includes URL converter)
self.drag_interceptor = DragInterceptor() self.drag_interceptor = DragInterceptor(config)
# Set up path validator
validator = PathValidator(config.allowed_roots)
self.drag_interceptor.set_validator(validator)
# Connect drag interceptor signals # Connect drag interceptor signals
self.drag_interceptor.drag_started.connect(self._on_drag_started) self.drag_interceptor.drag_started.connect(self._on_drag_started)
@ -283,6 +312,26 @@ class MainWindow(QMainWindow):
# Install the drag bridge script # Install the drag bridge script
self._install_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 # Set up central widget with layout
central_widget = QWidget() central_widget = QWidget()
layout = QVBoxLayout() layout = QVBoxLayout()
@ -353,19 +402,55 @@ class MainWindow(QMainWindow):
def _install_bridge_script(self) -> None: def _install_bridge_script(self) -> None:
"""Install the drag bridge JavaScript via QWebEngineScript. """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 = QWebEngineScript()
script.setName("webdrop-bridge") script.setName("webdrop-bridge")
script.setInjectionPoint(QWebEngineScript.InjectionPoint.DocumentReady) script.setInjectionPoint(QWebEngineScript.InjectionPoint.DocumentCreation)
script.setWorldId(QWebEngineScript.ScriptWorldId.MainWorld) script.setWorldId(QWebEngineScript.ScriptWorldId.MainWorld)
script.setRunsOnSubFrames(False) 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 # Load bridge script from file
script_path = Path(__file__).parent / "bridge_script.js" script_path = Path(__file__).parent / "bridge_script.js"
try: try:
with open(script_path, 'r', encoding='utf-8') as f: 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) self.web_view.page().scripts().insert(script)
logger.debug(f"Installed bridge script from {script_path}") logger.debug(f"Installed bridge script from {script_path}")
except (OSError, IOError) as e: except (OSError, IOError) as e:
@ -399,23 +484,248 @@ class MainWindow(QMainWindow):
# Silently fail if stylesheet can't be read # Silently fail if stylesheet can't be read
pass pass
def _on_drag_started(self, paths: list) -> None: def _on_drag_started(self, source: str, local_path: str) -> None:
"""Handle successful drag initiation. """Handle successful drag initiation.
Args: 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 logger.info(f"Drag started: {source} -> {local_path}")
pass # 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. """Handle drag operation failure.
Args: Args:
source: Original URL or path from web content
error: Error message error: Error message
""" """
# Can be extended with logging or user notification logger.warning(f"Drag failed for {source}: {error}")
pass # 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: def _create_navigation_toolbar(self) -> None:
"""Create navigation toolbar with Home, Back, Forward, Refresh buttons. """Create navigation toolbar with Home, Back, Forward, Refresh buttons.
@ -488,7 +798,7 @@ class MainWindow(QMainWindow):
Args: Args:
status: Status text to display status: Status text to display
emoji: Optional emoji prefix (🔄, , , ) emoji: Optional emoji prefix (rotating, checkmark, download, warning symbols)
""" """
if emoji: if emoji:
self.update_status_label.setText(f"{emoji} {status}") self.update_status_label.setText(f"{emoji} {status}")
@ -559,24 +869,11 @@ class MainWindow(QMainWindow):
# Can be extended with save operations or cleanup # Can be extended with save operations or cleanup
event.accept() 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: def check_for_updates_startup(self) -> None:
"""Check for updates on application startup. """Check for updates on application startup.
Runs asynchronously in background without blocking UI. 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 from webdrop_bridge.core.updater import UpdateManager

View file

@ -1,13 +1,106 @@
"""Restricted web view with URL whitelist enforcement for Kiosk-mode.""" """Restricted web view with URL whitelist enforcement for Kiosk-mode."""
import fnmatch 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.QtGui import QDesktopServices
from PySide6.QtWebEngineCore import QWebEngineNavigationRequest from PySide6.QtWebEngineCore import QWebEngineNavigationRequest, QWebEnginePage, QWebEngineProfile
from PySide6.QtWebEngineWidgets import QWebEngineView 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): class RestrictedWebEngineView(QWebEngineView):
"""Web view that enforces URL whitelist for Kiosk-mode security. """Web view that enforces URL whitelist for Kiosk-mode security.
@ -27,31 +120,81 @@ class RestrictedWebEngineView(QWebEngineView):
super().__init__() super().__init__()
self.allowed_urls = allowed_urls or [] 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 # Connect to navigation request handler
self.page().navigationRequested.connect(self._on_navigation_requested) self.page().navigationRequested.connect(self._on_navigation_requested)
def _on_navigation_requested( def _create_persistent_profile(self) -> QWebEngineProfile:
self, request: QWebEngineNavigationRequest """Create and configure a persistent web engine profile.
) -> None:
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. """Handle navigation requests and enforce URL whitelist.
Args: Args:
request: Navigation request to process request: Navigation request to process
""" """
url = request.url url = request.url()
# If no restrictions, allow all URLs # If no restrictions, allow all URLs
if not self.allowed_urls: if not self.allowed_urls:
return return
# Check if URL matches whitelist # 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) # Allow the navigation (default behavior)
return return
# URL not whitelisted - open in system browser # URL not whitelisted - open in system browser
request.reject() request.reject()
QDesktopServices.openUrl(url) # type: ignore[operator] QDesktopServices.openUrl(url)
def _is_url_allowed(self, url: QUrl) -> bool: def _is_url_allowed(self, url: QUrl) -> bool:
"""Check if a URL matches the whitelist. """Check if a URL matches the whitelist.
@ -98,4 +241,3 @@ class RestrictedWebEngineView(QWebEngineView):
return True return True
return False return False

95
test_download.html Normal file
View file

@ -0,0 +1,95 @@
<!DOCTYPE html>
<html lang="de">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>Download Test</title>
<style>
body {
font-family: Arial, sans-serif;
padding: 40px;
background: #f5f5f5;
}
.test-section {
background: white;
padding: 30px;
margin: 20px 0;
border-radius: 8px;
box-shadow: 0 2px 4px rgba(0,0,0,0.1);
}
h2 {
color: #333;
margin-top: 0;
}
a {
display: inline-block;
padding: 10px 20px;
background: #007bff;
color: white;
text-decoration: none;
border-radius: 4px;
margin: 10px 10px 10px 0;
}
a:hover {
background: #0056b3;
}
button {
padding: 10px 20px;
background: #28a745;
color: white;
border: none;
border-radius: 4px;
cursor: pointer;
margin: 10px 10px 10px 0;
}
button:hover {
background: #218838;
}
</style>
</head>
<body>
<h1>🧪 Download Test Seite</h1>
<div class="test-section">
<h2>Test 1: Direct Download Link</h2>
<p>Download einer Text-Datei via direktem Link:</p>
<a href="data:text/plain;charset=utf-8,Das ist eine Test-Datei%0AMit mehreren Zeilen%0AZum Testen" download="test.txt">📥 Download test.txt</a>
</div>
<div class="test-section">
<h2>Test 2: JavaScript Download (Blob)</h2>
<p>Download via JavaScript Blob:</p>
<button onclick="downloadBlob()">📥 Download blob.txt</button>
</div>
<div class="test-section">
<h2>Test 3: Base64 Image Download</h2>
<p>Download eines kleinen Bildes:</p>
<a href="data:image/png;base64,iVBORw0KGgoAAAANSUhEUgAAAAoAAAAKCAYAAACNMs+9AAAAFUlEQVR42mNk+M9Qz0AEYBxVSF+FABJADveWkH6oAAAAAElFTkSuQmCC" download="test.png">📥 Download test.png</a>
</div>
<div class="test-section">
<h2>Test 4: External Link (should open in browser)</h2>
<p>PDF von externem Server:</p>
<a href="https://www.w3.org/WAI/ER/tests/xhtml/testfiles/resources/pdf/dummy.pdf" target="_blank">📥 Download external PDF</a>
</div>
<script>
function downloadBlob() {
const content = 'Das ist eine Test-Datei\nErstellt via JavaScript Blob\n' + new Date().toISOString();
const blob = new Blob([content], { type: 'text/plain' });
const url = URL.createObjectURL(blob);
const a = document.createElement('a');
a.href = url;
a.download = 'blob_test.txt';
document.body.appendChild(a);
a.click();
document.body.removeChild(a);
URL.revokeObjectURL(url);
console.log('Download via Blob initiated');
}
</script>
</body>
</html>

19
test_download.py Normal file
View file

@ -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()}")

24
test_url_mappings.py Normal file
View file

@ -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}")

View file

@ -98,12 +98,13 @@ class TestConfigFromEnv:
Config.from_env(str(env_file)) Config.from_env(str(env_file))
def test_from_env_invalid_root_path(self, tmp_path): 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 = tmp_path / ".env"
env_file.write_text("ALLOWED_ROOTS=/nonexistent/path/that/does/not/exist\n") env_file.write_text("ALLOWED_ROOTS=/nonexistent/path/that/does/not/exist\n")
with pytest.raises(ConfigurationError, match="does not exist"): # Should not raise - just logs warning and returns empty allowed_roots
Config.from_env(str(env_file)) 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): def test_from_env_empty_webapp_url(self, tmp_path):
"""Test that empty webapp URL raises ConfigurationError.""" """Test that empty webapp URL raises ConfigurationError."""

View file

@ -3,63 +3,79 @@
from pathlib import Path from pathlib import Path
from unittest.mock import MagicMock, patch 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.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: class TestDragInterceptorInitialization:
"""Test DragInterceptor initialization and setup.""" """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.""" """Test DragInterceptor can be instantiated."""
interceptor = DragInterceptor() interceptor = DragInterceptor(test_config)
assert interceptor is not None 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.""" """Test DragInterceptor has required signals."""
interceptor = DragInterceptor() interceptor = DragInterceptor(test_config)
assert hasattr(interceptor, "drag_started") assert hasattr(interceptor, "drag_started")
assert hasattr(interceptor, "drag_failed") assert hasattr(interceptor, "drag_failed")
def test_set_validator(self, qtbot, tmp_path): def test_set_validator(self, qtbot, test_config):
"""Test setting validator on drag interceptor.""" """Test validator is set during construction."""
interceptor = DragInterceptor() interceptor = DragInterceptor(test_config)
validator = PathValidator([tmp_path]) assert interceptor._validator is not None
interceptor.set_validator(validator)
assert interceptor._validator is validator
class TestDragInterceptorValidation: class TestDragInterceptorValidation:
"""Test path validation in drag operations.""" """Test path validation in drag operations."""
def test_initiate_drag_no_files(self, qtbot): def test_handle_drag_empty_text(self, qtbot, test_config):
"""Test initiating drag with no files fails.""" """Test handling drag with empty text fails."""
interceptor = DragInterceptor() interceptor = DragInterceptor(test_config)
with qtbot.waitSignal(interceptor.drag_failed): with qtbot.waitSignal(interceptor.drag_failed):
result = interceptor.initiate_drag([]) result = interceptor.handle_drag("")
assert result is False assert result is False
def test_initiate_drag_no_validator(self, qtbot): def test_handle_drag_valid_file_path(self, qtbot, tmp_path):
"""Test initiating drag without validator fails.""" """Test handling drag with valid file path."""
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."""
# Create a test file # Create a test file
test_file = tmp_path / "test.txt" test_file = tmp_path / "test.txt"
test_file.write_text("test content") test_file.write_text("test content")
interceptor = DragInterceptor() config = Config(
validator = PathValidator([tmp_path]) app_name="Test",
interceptor.set_validator(validator) 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 # Mock the drag operation to simulate success
with patch("webdrop_bridge.core.drag_interceptor.QDrag") as mock_drag: 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_instance.exec.return_value = Qt.DropAction.CopyAction
mock_drag.return_value = mock_drag_instance 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 # Should return True on successful drag
assert result is True 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.""" """Test drag with invalid path fails."""
interceptor = DragInterceptor() interceptor = DragInterceptor(test_config)
validator = PathValidator([tmp_path])
interceptor.set_validator(validator)
# Path outside allowed roots # Path outside allowed roots
invalid_path = Path("/etc/passwd") invalid_path = "/etc/passwd"
with qtbot.waitSignal(interceptor.drag_failed): with qtbot.waitSignal(interceptor.drag_failed):
result = interceptor.initiate_drag([str(invalid_path)]) result = interceptor.handle_drag(invalid_path)
assert result is False 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.""" """Test drag with nonexistent file fails."""
interceptor = DragInterceptor() interceptor = DragInterceptor(test_config)
validator = PathValidator([tmp_path])
interceptor.set_validator(validator)
nonexistent = tmp_path / "nonexistent.txt" nonexistent = tmp_path / "nonexistent.txt"
with qtbot.waitSignal(interceptor.drag_failed): with qtbot.waitSignal(interceptor.drag_failed):
result = interceptor.initiate_drag([str(nonexistent)]) result = interceptor.handle_drag(str(nonexistent))
assert result is False assert result is False
class TestDragInterceptorMultipleFiles: class TestDragInterceptorAzureURL:
"""Test drag operations with multiple files.""" """Test Azure URL to local path conversion in drag operations."""
def test_initiate_drag_multiple_files(self, qtbot, tmp_path): def test_handle_drag_azure_url(self, qtbot, tmp_path):
"""Test drag with multiple valid files.""" """Test handling drag with Azure Blob Storage URL."""
# Create test files from webdrop_bridge.config import URLMapping
file1 = tmp_path / "file1.txt"
file2 = tmp_path / "file2.txt"
file1.write_text("content 1")
file2.write_text("content 2")
interceptor = DragInterceptor() # Create test file that would be the result
validator = PathValidator([tmp_path]) test_file = tmp_path / "test.png"
interceptor.set_validator(validator) 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: with patch("webdrop_bridge.core.drag_interceptor.QDrag") as mock_drag:
mock_drag_instance = MagicMock() mock_drag_instance = MagicMock()
from PySide6.QtCore import Qt
mock_drag_instance.exec.return_value = Qt.DropAction.CopyAction mock_drag_instance.exec.return_value = Qt.DropAction.CopyAction
mock_drag.return_value = mock_drag_instance 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 assert result is True
def test_initiate_drag_mixed_valid_invalid(self, qtbot, tmp_path): def test_handle_drag_unmapped_url(self, qtbot, test_config):
"""Test drag with mix of valid and invalid paths fails.""" """Test handling drag with unmapped URL fails."""
test_file = tmp_path / "valid.txt" interceptor = DragInterceptor(test_config)
test_file.write_text("content")
interceptor = DragInterceptor() # URL with no mapping
validator = PathValidator([tmp_path]) unmapped_url = "https://unknown.blob.core.windows.net/container/file.png"
interceptor.set_validator(validator)
# Mix of valid and invalid paths
with qtbot.waitSignal(interceptor.drag_failed): with qtbot.waitSignal(interceptor.drag_failed):
result = interceptor.initiate_drag( result = interceptor.handle_drag(unmapped_url)
[str(test_file), "/etc/passwd"]
)
assert result is False 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: class TestDragInterceptorSignals:
"""Test signal emission on drag operations.""" """Test signal emission on drag operations."""
@ -185,153 +178,60 @@ class TestDragInterceptorSignals:
test_file = tmp_path / "test.txt" test_file = tmp_path / "test.txt"
test_file.write_text("content") test_file.write_text("content")
interceptor = DragInterceptor() config = Config(
validator = PathValidator([tmp_path]) app_name="Test",
interceptor.set_validator(validator) app_version="1.0.0",
log_level="INFO",
from PySide6.QtCore import Qt 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 # Connect to signal manually
signal_spy = [] 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: with patch("webdrop_bridge.core.drag_interceptor.QDrag") as mock_drag:
mock_drag_instance = MagicMock() mock_drag_instance = MagicMock()
mock_drag_instance.exec.return_value = Qt.DropAction.CopyAction mock_drag_instance.exec.return_value = Qt.DropAction.CopyAction
mock_drag.return_value = mock_drag_instance 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 # Verify result and signal emission
assert result is True assert result is True
assert len(signal_spy) == 1 assert len(signal_spy) == 1
def test_drag_failed_signal_on_no_files(self, qtbot): def test_drag_failed_signal_on_empty_text(self, qtbot, test_config):
"""Test drag_failed signal on empty file list.""" """Test drag_failed signal on empty text."""
interceptor = DragInterceptor() interceptor = DragInterceptor(test_config)
# Connect to signal manually # Connect to signal manually
signal_spy = [] 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 # Verify result and signal emission
assert result is False assert result is False
assert len(signal_spy) == 1 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.""" """Test drag_failed signal on validation failure."""
interceptor = DragInterceptor() interceptor = DragInterceptor(test_config)
validator = PathValidator([tmp_path])
interceptor.set_validator(validator)
# Connect to signal manually # Connect to signal manually
signal_spy = [] 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 # Verify result and signal emission
assert result is False assert result is False
assert len(signal_spy) == 1 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

View file

@ -231,10 +231,12 @@ class TestMainWindowDragIntegration:
assert window.drag_interceptor.drag_started is not None assert window.drag_interceptor.drag_started is not None
assert window.drag_interceptor.drag_failed 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 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) window = MainWindow(sample_config)
qtbot.addWidget(window) qtbot.addWidget(window)
@ -243,29 +245,32 @@ class TestMainWindowDragIntegration:
test_file.write_text("test") test_file.write_text("test")
with patch.object( with patch.object(
window.drag_interceptor, "initiate_drag" window.drag_interceptor, "handle_drag"
) as mock_drag: ) as mock_drag:
mock_drag.return_value = True mock_drag.return_value = True
result = window.initiate_drag([str(test_file)]) # Call through bridge
window._drag_bridge.start_file_drag(str(test_file))
mock_drag.assert_called_once_with([str(test_file)]) # Process deferred QTimer.singleShot(0, ...) call
assert result is True QCoreApplication.processEvents()
mock_drag.assert_called_once_with(str(test_file))
def test_on_drag_started_called(self, qtbot, sample_config): def test_on_drag_started_called(self, qtbot, sample_config):
"""Test _on_drag_started handler can be called.""" """Test _on_drag_started handler can be called."""
window = MainWindow(sample_config) window = MainWindow(sample_config)
qtbot.addWidget(window) qtbot.addWidget(window)
# Should not raise # Should not raise - new signature has source and local_path
window._on_drag_started(["/some/path"]) window._on_drag_started("https://example.com/file.png", "/local/path/file.png")
def test_on_drag_failed_called(self, qtbot, sample_config): def test_on_drag_failed_called(self, qtbot, sample_config):
"""Test _on_drag_failed handler can be called.""" """Test _on_drag_failed handler can be called."""
window = MainWindow(sample_config) window = MainWindow(sample_config)
qtbot.addWidget(window) qtbot.addWidget(window)
# Should not raise # Should not raise - new signature has source and error
window._on_drag_failed("Test error message") window._on_drag_failed("https://example.com/file.png", "Test error message")
class TestMainWindowURLWhitelist: class TestMainWindowURLWhitelist:

View file

@ -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

View file

@ -22,11 +22,12 @@ class TestPathValidator:
assert len(validator.allowed_roots) == 2 assert len(validator.allowed_roots) == 2
def test_validator_nonexistent_root(self, tmp_path): 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" nonexistent = tmp_path / "nonexistent"
with pytest.raises(ValidationError, match="does not exist"): # Should not raise - just logs warning and skips the root
PathValidator([nonexistent]) validator = PathValidator([nonexistent])
assert len(validator.allowed_roots) == 0 # Non-existent roots are skipped
def test_validator_non_directory_root(self, tmp_path): def test_validator_non_directory_root(self, tmp_path):
"""Test that non-directory root raises ValidationError.""" """Test that non-directory root raises ValidationError."""

View file

@ -162,20 +162,26 @@
<div class="drag-items"> <div class="drag-items">
<div class="drag-item" draggable="true" id="dragItem1"> <div class="drag-item" draggable="true" id="dragItem1">
<div class="icon">🖼️</div> <div class="icon">🖼️</div>
<h3>Sample Image</h3> <h3>Local Z:\ Image</h3>
<p id="path1">Z:\data\test-image.jpg</p> <p id="path1">Z:\data\test-image.jpg</p>
</div> </div>
<div class="drag-item" draggable="true" id="dragItem2"> <div class="drag-item" draggable="true" id="dragItem2">
<div class="icon">📄</div> <div class="icon">📄</div>
<h3>Sample Document</h3> <h3>Local Z:\ Document</h3>
<p id="path2">Z:\data\API_DOCUMENTATION.pdf</p> <p id="path2">Z:\data\API_DOCUMENTATION.pdf</p>
</div> </div>
<div class="drag-item" draggable="true" id="dragItem3"> <div class="drag-item" draggable="true" id="dragItem3">
<div class="icon">📊</div> <div class="icon">☁️</div>
<h3>Sample Data</h3> <h3>Azure Blob Storage Image</h3>
<p id="path3">C:\Users\Public\data.csv</p> <p id="path3">https://wpsagravitystg.file.core.windows.net/wpsagravitysync/aN5PysnXIuRECzcRbvHkjL7g0/Hintergrund_Agravity.png</p>
</div>
<div class="drag-item" draggable="true" id="dragItem4">
<div class="icon">☁️</div>
<h3>Azure Blob Storage Document</h3>
<p id="path4">https://wpsagravitystg.file.core.windows.net/wpsagravitysync/test/document.pdf</p>
</div> </div>
</div> </div>
@ -183,15 +189,59 @@
<h4>How to test:</h4> <h4>How to test:</h4>
<ol> <ol>
<li>Open InDesign, Word, or Notepad++</li> <li>Open InDesign, Word, or Notepad++</li>
<li>Drag one of the items below to the application</li> <li>Drag one of the items above to the application</li>
<li>The file path should be converted to a real file drag</li> <li>Local Z:\ paths and Azure URLs will be converted to file drags</li>
<li>Azure URLs will be mapped to Z:\ paths automatically</li>
<li>Check the browser console (F12) for debug info</li> <li>Check the browser console (F12) for debug info</li>
</ol> </ol>
<p><strong>Note:</strong> When dragging images from web apps like Agravity, the browser may not provide text/plain data. Press <kbd>ALT</kbd> while dragging to force text drag mode.</p>
</div> </div>
<div class="footer"> <div class="footer">
<p>WebDrop Bridge v1.0.0 | Built with Qt and PySide6</p> <p>WebDrop Bridge v1.0.0 | Built with Qt and PySide6</p>
</div> </div>
</div> </div>
<script>
// Debug logging for drag operations
document.addEventListener('dragstart', function(e) {
var statusEl = document.getElementById('statusMessage');
// Log what's being dragged
var dt = e.dataTransfer;
var path = dt.getData('text/plain') || dt.getData('text/uri-list');
if (!path && e.target.tagName === 'IMG') {
path = e.target.src;
} else if (!path && e.target.tagName === 'A') {
path = e.target.href;
} else if (!path) {
var card = e.target.closest('.drag-item');
if (card) {
var pathEl = card.querySelector('p');
if (pathEl) path = pathEl.textContent.trim();
}
}
if (path) {
statusEl.className = 'status-message info';
statusEl.textContent = 'Dragging: ' + path;
console.log('[WebDrop] Drag started:', path);
}
}, false);
document.addEventListener('dragend', function(e) {
var statusEl = document.getElementById('statusMessage');
statusEl.className = 'status-message success';
statusEl.textContent = 'Drag completed!';
console.log('[WebDrop] Drag ended');
// Reset after 2 seconds
setTimeout(function() {
statusEl.className = 'status-message info';
statusEl.textContent = 'Ready to test drag and drop';
}, 2000);
}, false);
</script>
</body> </body>
</html> </html>