Coverage for integrations / remote_desktop / drag_drop.py: 69.9%

1""" 

2Drag-and-Drop Bridge — Cross-device file drag-and-drop over HARTOS transport. 

3 

4Local→Remote: detect drag onto viewer → DLP scan → transfer → simulate drop at (x,y) 

5Remote→Local: receive drop event → transfer → place file locally 

6 

7Composes existing modules: 

8 - file_transfer.py: FileTransfer.send_file() for actual transfer 

9 - security.py: scan_file_transfer() for DLP 

10 - input_handler.py: InputHandler for drop simulation on remote 

11 

12RustDesk already has native drag-drop; this provides the same for the native 

13HARTOS transport fallback. 

14""" 

15 

16import logging 

17import os 

18import platform 

19import subprocess 

20import threading 

21import time 

22from dataclasses import dataclass, field 

23from enum import Enum 

24from typing import Callable, Dict, List, Optional 

25 

26logger = logging.getLogger('hevolve.remote_desktop') 

27 

28 

29class DragDropState(Enum): 

30 IDLE = 'idle' 

31 DRAG_STARTED = 'drag_started' 

32 TRANSFERRING = 'transferring' 

33 DROP_PENDING = 'drop_pending' 

34 COMPLETED = 'completed' 

35 ERROR = 'error' 

36 

37 

38@dataclass 

39class DragDropEvent: 

40 """Cross-device drag-and-drop event.""" 

41 direction: str # 'local_to_remote' or 'remote_to_local' 

42 file_paths: List[str] = field(default_factory=list) 

43 drop_x: int = 0 

44 drop_y: int = 0 

45 state: DragDropState = DragDropState.IDLE 

46 error: Optional[str] = None 

47 bytes_transferred: int = 0 

48 

49 def to_dict(self) -> dict: 

50 return { 

51 'direction': self.direction, 

52 'file_paths': self.file_paths, 

53 'drop_x': self.drop_x, 

54 'drop_y': self.drop_y, 

55 'state': self.state.value, 

56 'error': self.error, 

57 'bytes_transferred': self.bytes_transferred, 

58 } 

59 

60 

61class DragDropBridge: 

62 """Cross-device drag-and-drop over HARTOS transport. 

63 

64 Orchestrates file transfer + input simulation for seamless drag-drop 

65 across local and remote devices. 

66 """ 

67 

68 def __init__(self, transport=None, input_handler=None): 

69 self._transport = transport 

70 self._input_handler = input_handler 

71 self._monitoring = False 

72 self._progress_callbacks: List[Callable] = [] 

73 self._active_drops: Dict[str, DragDropEvent] = {} 

74 self._lock = threading.Lock() 

75 self._receive_dir: Optional[str] = None 

76 

77 def start_monitoring(self) -> bool: 

78 """Start monitoring for OS drag events entering the viewer window. 

79 

80 Returns True if monitoring started successfully. 

81 """ 

82 self._monitoring = True 

83 logger.info("Drag-and-drop bridge monitoring started") 

84 return True 

85 

86 def stop_monitoring(self) -> None: 

87 """Stop monitoring for drag events.""" 

88 self._monitoring = False 

89 logger.info("Drag-and-drop bridge monitoring stopped") 

90 

91 @property 

92 def is_monitoring(self) -> bool: 

93 return self._monitoring 

94 

95 def set_receive_directory(self, path: str) -> None: 

96 """Set directory for received files.""" 

97 self._receive_dir = path 

98 

99 def handle_local_drop(self, file_paths: List[str], 

100 x: int, y: int) -> dict: 

101 """Handle files dropped onto the viewer → transfer + remote drop. 

102 

103 1. Validate file paths exist 

104 2. DLP scan each file 

105 3. Transfer via FileTransfer 

106 4. Send drop event to remote host 

107 5. Host simulates file drop at (x, y) 

108 

109 Args: 

110 file_paths: Local file paths to send. 

111 x: Drop X coordinate on remote screen. 

112 y: Drop Y coordinate on remote screen. 

113 

114 Returns: 

115 {success, files_sent, errors} 

116 """ 

117 event = DragDropEvent( 

118 direction='local_to_remote', 

119 file_paths=file_paths, 

120 drop_x=x, 

121 drop_y=y, 

122 state=DragDropState.DRAG_STARTED, 

123 ) 

124 self._notify_progress(event) 

125 

126 # Validate paths 

127 valid_paths = [] 

128 errors = [] 

129 for path in file_paths: 

130 if os.path.exists(path): 

131 valid_paths.append(path) 

132 else: 

133 errors.append(f"File not found: {path}") 

134 

135 if not valid_paths: 

136 event.state = DragDropState.ERROR 

137 event.error = 'No valid files' 

138 self._notify_progress(event) 

139 return {'success': False, 'error': 'No valid files', 'errors': errors} 

140 

141 # DLP scan 

142 event.state = DragDropState.TRANSFERRING 

143 self._notify_progress(event) 

144 

145 blocked = [] 

146 for path in valid_paths: 

147 try: 

148 from integrations.remote_desktop.security import scan_file_transfer 

149 allowed, reason = scan_file_transfer(os.path.basename(path)) 

150 if not allowed: 

151 blocked.append(f"DLP blocked {os.path.basename(path)}: {reason}") 

152 except Exception: 

153 pass # If DLP unavailable, allow 

154 

155 if blocked: 

156 errors.extend(blocked) 

157 valid_paths = [p for p in valid_paths 

158 if os.path.basename(p) not in 

159 ' '.join(blocked)] 

160 

161 # Transfer files 

162 results = [] 

163 total_bytes = 0 

164 for path in valid_paths: 

165 result = self._transfer_file(path) 

166 results.append(result) 

167 if result.get('success'): 

168 total_bytes += result.get('bytes_sent', 0) 

169 else: 

170 errors.append(result.get('error', 'Transfer failed')) 

171 

172 event.bytes_transferred = total_bytes 

173 files_sent = sum(1 for r in results if r.get('success')) 

174 

175 # Send drop event to remote 

176 if files_sent > 0 and self._transport: 

177 try: 

178 filenames = [os.path.basename(p) for p in valid_paths[:files_sent]] 

179 self._transport.send_event({ 

180 'type': 'drag_drop', 

181 'action': 'DROP', 

182 'files': filenames, 

183 'x': x, 

184 'y': y, 

185 'direction': 'local_to_remote', 

186 }) 

187 except Exception as e: 

188 errors.append(f"Drop event send failed: {e}") 

189 

190 event.state = (DragDropState.COMPLETED if files_sent > 0 

191 else DragDropState.ERROR) 

192 event.error = '; '.join(errors) if errors else None 

193 self._notify_progress(event) 

194 

195 return { 

196 'success': files_sent > 0, 

197 'files_sent': files_sent, 

198 'total_files': len(file_paths), 

199 'bytes_transferred': total_bytes, 

200 'errors': errors, 

201 } 

202 

203 def handle_remote_drop(self, event: dict) -> dict: 

204 """Handle drop event received from remote → receive files locally. 

205 

206 event: {type: 'drag_drop', action: 'DROP', files: [...], x, y} 

207 

208 Returns: 

209 {success, files_received, save_dir} 

210 """ 

211 action = event.get('action', '') 

212 if action != 'DROP': 

213 return {'success': False, 'error': f'Unknown action: {action}'} 

214 

215 files = event.get('files', []) 

216 x = event.get('x', 0) 

217 y = event.get('y', 0) 

218 

219 save_dir = self._receive_dir or self._get_default_receive_dir() 

220 

221 drop_event = DragDropEvent( 

222 direction='remote_to_local', 

223 file_paths=files, 

224 drop_x=x, 

225 drop_y=y, 

226 state=DragDropState.DROP_PENDING, 

227 ) 

228 self._notify_progress(drop_event) 

229 

230 # Files should have been transferred via FileTransfer already 

231 # Just acknowledge the drop 

232 drop_event.state = DragDropState.COMPLETED 

233 self._notify_progress(drop_event) 

234 

235 return { 

236 'success': True, 

237 'files_received': len(files), 

238 'save_dir': save_dir, 

239 'x': x, 

240 'y': y, 

241 } 

242 

243 def handle_file_drop_on_host(self, event: dict) -> dict: 

244 """Host-side: simulate a file drop at cursor position. 

245 

246 Called when remote viewer sends files + drop position. Uses InputHandler 

247 or platform-specific file placement. 

248 """ 

249 x = event.get('x', 0) 

250 y = event.get('y', 0) 

251 files = event.get('files', []) 

252 

253 if self._input_handler: 

254 # Move cursor to drop position 

255 try: 

256 self._input_handler.handle_input_event({ 

257 'type': 'move', 

258 'x': x, 

259 'y': y, 

260 }) 

261 except Exception: 

262 pass 

263 

264 # Open file explorer at the received file location 

265 save_dir = self._receive_dir or self._get_default_receive_dir() 

266 self._open_file_location(save_dir) 

267 

268 return { 

269 'success': True, 

270 'files': files, 

271 'position': {'x': x, 'y': y}, 

272 'save_dir': save_dir, 

273 } 

274 

275 def on_progress(self, callback: Callable[[DragDropEvent], None]) -> None: 

276 """Register a progress callback.""" 

277 self._progress_callbacks.append(callback) 

278 

279 def get_status(self) -> dict: 

280 """Get drag-drop bridge status.""" 

281 return { 

282 'monitoring': self._monitoring, 

283 'has_transport': self._transport is not None, 

284 'has_input_handler': self._input_handler is not None, 

285 'receive_dir': self._receive_dir, 

286 } 

287 

288 # ── Internal helpers ────────────────────────────────────── 

289 

290 def _transfer_file(self, local_path: str) -> dict: 

291 """Transfer a single file via FileTransfer module.""" 

292 if not self._transport: 

293 return {'success': False, 'error': 'No transport'} 

294 

295 try: 

296 from integrations.remote_desktop.file_transfer import FileTransfer 

297 ft = FileTransfer() 

298 return ft.send_file(self._transport, local_path) 

299 except Exception as e: 

300 return {'success': False, 'error': str(e)} 

301 

302 def _notify_progress(self, event: DragDropEvent) -> None: 

303 """Notify all progress callbacks.""" 

304 for cb in self._progress_callbacks: 

305 try: 

306 cb(event) 

307 except Exception: 

308 pass 

309 

310 def _get_default_receive_dir(self) -> str: 

311 """Get default directory for received files.""" 

312 # Try standard Downloads folder 

313 home = os.path.expanduser('~') 

314 downloads = os.path.join(home, 'Downloads') 

315 if os.path.isdir(downloads): 

316 return downloads 

317 return home 

318 

319 def _open_file_location(self, path: str) -> None: 

320 """Open file manager at the given path (platform-specific).""" 

321 system = platform.system() 

322 try: 

323 if system == 'Windows': 

324 subprocess.Popen(['explorer', path]) 

325 elif system == 'Darwin': 

326 subprocess.Popen(['open', path]) 

327 elif system == 'Linux': 

328 subprocess.Popen(['xdg-open', path]) 

329 except Exception as e: 

330 logger.debug(f"Could not open file location: {e}")