Coverage for integrations / remote_desktop / clipboard_sync.py: 61.2%

1""" 

2Clipboard Sync — Bidirectional clipboard sharing between host and viewer. 

3 

4Reuses pyperclip (already in codebase via local_computer_tool.py:28). 

5DLP scan on outbound clipboard via security/dlp_engine.py. 

6""" 

7 

8import logging 

9import threading 

10import time 

11from typing import Callable, Optional 

12 

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

14 

15# ── Optional dependency ───────────────────────────────────────── 

16 

17_pyperclip = None 

18try: 

19 import pyperclip as _pyperclip_module 

20 _pyperclip = _pyperclip_module 

21except ImportError: 

22 pass 

23 

24 

25class ClipboardSync: 

26 """Bidirectional clipboard synchronization. 

27 

28 Monitors local clipboard for changes and sends to remote. 

29 Receives remote clipboard content and applies locally. 

30 """ 

31 

32 POLL_INTERVAL = 0.5 # 500ms 

33 

34 def __init__(self, on_change: Optional[Callable[[str], None]] = None, 

35 dlp_enabled: bool = True): 

36 """ 

37 Args: 

38 on_change: Callback when local clipboard changes (sends to remote) 

39 dlp_enabled: Whether to DLP-scan outbound clipboard 

40 """ 

41 self._on_change = on_change 

42 self._dlp_enabled = dlp_enabled 

43 self._last_content: Optional[str] = None 

44 self._running = False 

45 self._thread: Optional[threading.Thread] = None 

46 self._lock = threading.Lock() 

47 self._paused = False 

48 

49 def start_monitoring(self) -> bool: 

50 """Start background clipboard monitoring thread. 

51 

52 Returns: 

53 True if started, False if pyperclip unavailable. 

54 """ 

55 if not _pyperclip: 

56 logger.warning("Clipboard sync unavailable: pyperclip not installed") 

57 return False 

58 if self._running: 

59 return True 

60 

61 self._running = True 

62 try: 

63 self._last_content = _pyperclip.paste() 

64 except Exception: 

65 self._last_content = '' 

66 

67 self._thread = threading.Thread( 

68 target=self._monitor_loop, 

69 daemon=True, 

70 name='clipboard-sync', 

71 ) 

72 self._thread.start() 

73 logger.info("Clipboard monitoring started") 

74 return True 

75 

76 def stop_monitoring(self) -> None: 

77 """Stop clipboard monitoring.""" 

78 self._running = False 

79 if self._thread: 

80 self._thread.join(timeout=2) 

81 self._thread = None 

82 logger.info("Clipboard monitoring stopped") 

83 

84 def pause(self) -> None: 

85 """Pause monitoring (e.g., while applying remote clipboard).""" 

86 self._paused = True 

87 

88 def resume(self) -> None: 

89 """Resume monitoring.""" 

90 self._paused = False 

91 

92 def apply_remote_clipboard(self, content: str) -> bool: 

93 """Set local clipboard from remote content. 

94 

95 Pauses monitoring to avoid echo loop. 

96 

97 Returns: 

98 True if clipboard was set successfully. 

99 """ 

100 if not _pyperclip: 

101 return False 

102 

103 with self._lock: 

104 self.pause() 

105 try: 

106 _pyperclip.copy(content) 

107 self._last_content = content 

108 logger.debug(f"Remote clipboard applied ({len(content)} chars)") 

109 return True 

110 except Exception as e: 

111 logger.warning(f"Failed to set clipboard: {e}") 

112 return False 

113 finally: 

114 self.resume() 

115 

116 def get_current(self) -> Optional[str]: 

117 """Get current clipboard content.""" 

118 if not _pyperclip: 

119 return None 

120 try: 

121 return _pyperclip.paste() 

122 except Exception: 

123 return None 

124 

125 def _monitor_loop(self) -> None: 

126 """Background loop: detect clipboard changes and notify.""" 

127 while self._running: 

128 try: 

129 if not self._paused: 

130 current = _pyperclip.paste() 

131 if current != self._last_content: 

132 self._last_content = current 

133 self._handle_change(current) 

134 except Exception as e: 

135 logger.debug(f"Clipboard poll error: {e}") 

136 

137 time.sleep(self.POLL_INTERVAL) 

138 

139 def _handle_change(self, content: str) -> None: 

140 """Process clipboard change — DLP scan + notify.""" 

141 if not content: 

142 return 

143 

144 # DLP scan outbound clipboard 

145 if self._dlp_enabled: 

146 try: 

147 from integrations.remote_desktop.security import scan_clipboard 

148 allowed, reason = scan_clipboard(content) 

149 if not allowed: 

150 logger.warning(f"Clipboard blocked by DLP: {reason}") 

151 return 

152 except Exception: 

153 pass 

154 

155 # Notify callback 

156 if self._on_change: 

157 try: 

158 self._on_change(content) 

159 except Exception as e: 

160 logger.error(f"Clipboard change callback error: {e}") 

161 

162 @property 

163 def is_running(self) -> bool: 

164 return self._running 

165 

166 def get_stats(self) -> dict: 

167 return { 

168 'running': self._running, 

169 'paused': self._paused, 

170 'dlp_enabled': self._dlp_enabled, 

171 'pyperclip_available': _pyperclip is not None, 

172 }