""" Mini serveur HTTP sur l'agent Windows pour les captures d'ecran a la demande et les operations fichiers. Ecoute sur le port 5006 (configurable via RPA_CAPTURE_PORT). Endpoints : GET /capture -> screenshot frais en base64 (JPEG) GET /health -> {"status": "ok"} POST /file-action -> operations fichiers (list, create, move, copy, sort) """ import threading import logging import json import base64 import io import os import time from http.server import HTTPServer, BaseHTTPRequestHandler logger = logging.getLogger(__name__) CAPTURE_PORT = int(os.environ.get("RPA_CAPTURE_PORT", "5006")) # Floutage des données sensibles (conformité AI Act) BLUR_SENSITIVE = os.environ.get("RPA_BLUR_SENSITIVE", "true").lower() in ("true", "1", "yes") class CaptureHandler(BaseHTTPRequestHandler): """Retourne un screenshot frais a chaque requete GET /capture. Gere aussi les actions fichiers via POST /file-action. """ def do_GET(self): if self.path == "/capture": self._handle_capture() elif self.path == "/health": self._send_json(200, {"status": "ok"}) else: self._send_json(404, {"error": "not found"}) def do_POST(self): if self.path == "/file-action": self._handle_file_action() else: self._send_json(404, {"error": "not found"}) def do_OPTIONS(self): """Gestion CORS preflight.""" self.send_response(200) self._cors_headers() self.send_header("Content-Length", "0") self.end_headers() # ------------------------------------------------------------------ def _handle_file_action(self): """Execute une action fichier sur la machine Windows locale. Body JSON attendu : {"action": "file_sort_by_ext", "params": {"source_dir": "C:\\..."}} """ try: content_length = int(self.headers.get("Content-Length", 0)) body = self.rfile.read(content_length) data = json.loads(body.decode("utf-8")) action = data.get("action", "") params = data.get("params", {}) if not action: self._send_json(400, {"error": "Parametre 'action' requis"}) return handler = _FileActionHandlerLocal() result = handler.execute(action, params) code = 500 if "error" in result else 200 self._send_json(code, result) except json.JSONDecodeError: self._send_json(400, {"error": "JSON invalide"}) except Exception as e: logger.error(f"Erreur file-action : {e}") self._send_json(500, {"error": str(e)}) # ------------------------------------------------------------------ def _handle_capture(self): """Capture l'ecran principal et le renvoie en base64 JPEG.""" t0 = time.perf_counter() try: import mss from PIL import Image with mss.mss() as sct: monitor = sct.monitors[1] # ecran principal raw = sct.grab(monitor) img = Image.frombytes("RGB", raw.size, raw.bgra, "raw", "BGRX") # Floutage des données sensibles (conformité AI Act) if BLUR_SENSITIVE: try: from ..vision.blur_sensitive import blur_sensitive_regions blur_sensitive_regions(img) except ImportError: logger.warning("Module blur_sensitive non disponible") buf = io.BytesIO() img.save(buf, format="JPEG", quality=80) img_b64 = base64.b64encode(buf.getvalue()).decode() elapsed_ms = (time.perf_counter() - t0) * 1000 logger.info(f"Capture {img.width}x{img.height} en {elapsed_ms:.0f}ms") self._send_json(200, { "image": img_b64, "width": img.width, "height": img.height, "format": "jpeg", "source": "windows_live", "capture_ms": round(elapsed_ms), }) except Exception as e: logger.error(f"Erreur capture : {e}") self._send_json(500, {"error": str(e)}) # ------------------------------------------------------------------ def _send_json(self, code: int, data: dict): body = json.dumps(data).encode() self.send_response(code) self.send_header("Content-Type", "application/json") self._cors_headers() self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) def _cors_headers(self): self.send_header("Access-Control-Allow-Origin", "*") self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS") self.send_header("Access-Control-Allow-Headers", "Content-Type") def log_message(self, format, *args): """Supprime les logs HTTP par defaut (trop verbeux).""" pass # --------------------------------------------------------------------------- # Gestionnaire d'actions fichiers local (execute sur la machine Windows) # --------------------------------------------------------------------------- # Repertoires autorises sur Windows (securite anti-traversal) _WIN_ALLOWED_ROOTS = [ "C:\\Users", "D:\\", "E:\\", ] def _normalize_win_path(path_str: str) -> str: """Normalise un chemin Windows.""" import ntpath return ntpath.normpath(path_str) def _is_safe_win_path(path_str: str) -> bool: """Verifie qu'un chemin Windows est dans une zone autorisee.""" if not path_str or not path_str.strip(): return False norm = _normalize_win_path(path_str).upper() return any(norm.startswith(root.upper()) for root in _WIN_ALLOWED_ROOTS) class _FileActionHandlerLocal: """Execute les operations fichiers sur la machine locale (Windows).""" def execute(self, action_type: str, params: dict) -> dict: """Dispatch vers la bonne methode selon le type d'action.""" handlers = { "file_list_dir": self._list_dir, "file_create_dir": self._create_dir, "file_move": self._move_file, "file_copy": self._copy_file, "file_sort_by_ext": self._sort_by_extension, } handler = handlers.get(action_type) if not handler: return {"error": f"Action fichier inconnue : {action_type}"} try: return handler(params) except Exception as e: logger.error(f"Erreur action fichier '{action_type}' : {e}") return {"error": str(e)} def _list_dir(self, params: dict) -> dict: """Liste les fichiers d'un dossier.""" import fnmatch as _fnmatch from pathlib import Path as _Path path_str = params.get("path", "") pattern = params.get("pattern", "*") if not path_str: return {"error": "Parametre 'path' requis"} if not _is_safe_win_path(path_str): return {"error": f"Chemin non autorise : {path_str}"} source = _Path(path_str) if not source.exists(): return {"error": f"Dossier introuvable : {path_str}"} if not source.is_dir(): return {"error": f"Pas un dossier : {path_str}"} files = [] extensions = {} for item in source.iterdir(): if item.is_file() and _fnmatch.fnmatch(item.name, pattern): ext = item.suffix.lstrip(".").lower() or "sans_extension" files.append({ "name": item.name, "extension": ext, "size": item.stat().st_size, "path": str(item), }) extensions[ext] = extensions.get(ext, 0) + 1 logger.info(f"Liste dossier '{path_str}' : {len(files)} fichiers") return {"files": files, "count": len(files), "extensions": extensions, "path": path_str} def _create_dir(self, params: dict) -> dict: """Cree un dossier (parents inclus).""" from pathlib import Path as _Path path_str = params.get("path", "") if not path_str: return {"error": "Parametre 'path' requis"} if not _is_safe_win_path(path_str): return {"error": f"Chemin non autorise : {path_str}"} target = _Path(path_str) existed = target.exists() target.mkdir(parents=True, exist_ok=True) logger.info(f"Dossier '{path_str}' {'existait deja' if existed else 'cree'}") return {"created": not existed, "path": path_str, "already_existed": existed} def _move_file(self, params: dict) -> dict: """Deplace ou renomme un fichier.""" import shutil as _shutil from pathlib import Path as _Path src = params.get("source", "") dst = params.get("destination", "") if not src or not dst: return {"error": "Parametres 'source' et 'destination' requis"} if not _is_safe_win_path(src): return {"error": f"Source non autorisee : {src}"} if not _is_safe_win_path(dst): return {"error": f"Destination non autorisee : {dst}"} if not _Path(src).exists(): return {"error": f"Fichier source introuvable : {src}"} _Path(dst).parent.mkdir(parents=True, exist_ok=True) _shutil.move(src, dst) logger.info(f"Fichier deplace : '{src}' -> '{dst}'") return {"moved": True, "source": src, "destination": dst} def _copy_file(self, params: dict) -> dict: """Copie un fichier.""" import shutil as _shutil from pathlib import Path as _Path src = params.get("source", "") dst = params.get("destination", "") if not src or not dst: return {"error": "Parametres 'source' et 'destination' requis"} if not _is_safe_win_path(src): return {"error": f"Source non autorisee : {src}"} if not _is_safe_win_path(dst): return {"error": f"Destination non autorisee : {dst}"} source = _Path(src) if not source.exists(): return {"error": f"Fichier source introuvable : {src}"} _Path(dst).parent.mkdir(parents=True, exist_ok=True) if source.is_dir(): _shutil.copytree(src, dst) else: _shutil.copy2(src, dst) logger.info(f"Fichier copie : '{src}' -> '{dst}'") return {"copied": True, "source": src, "destination": dst} def _sort_by_extension(self, params: dict) -> dict: """Classe les fichiers par extension dans des sous-dossiers.""" import shutil as _shutil from pathlib import Path as _Path source_dir_str = params.get("source_dir", "") create_subdirs = params.get("create_subdirs", True) if not source_dir_str: return {"error": "Parametre 'source_dir' requis"} if not _is_safe_win_path(source_dir_str): return {"error": f"Chemin non autorise : {source_dir_str}"} source = _Path(source_dir_str) if not source.exists(): return {"error": f"Dossier introuvable : {source_dir_str}"} if not source.is_dir(): return {"error": f"Pas un dossier : {source_dir_str}"} moved = [] extensions = {} for f in source.iterdir(): if f.is_file(): ext = f.suffix.lstrip(".").lower() or "sans_extension" target_dir = source / ext if create_subdirs: target_dir.mkdir(exist_ok=True) elif not target_dir.exists(): continue dest = target_dir / f.name # Eviter ecrasement if dest.exists(): base = f.stem counter = 1 while dest.exists(): dest = target_dir / f"{base}_{counter}{f.suffix}" counter += 1 _shutil.move(str(f), str(dest)) moved.append({"file": f.name, "to": ext, "destination": str(dest)}) extensions[ext] = extensions.get(ext, 0) + 1 logger.info( f"Classement par extension '{source_dir_str}' : {len(moved)} fichiers" ) return { "moved": moved, "count": len(moved), "extensions": extensions, "source_dir": source_dir_str, } class CaptureServer: """Serveur de capture d'ecran en temps reel (thread daemon).""" def __init__(self, port: int = CAPTURE_PORT): self._port = port self._server: HTTPServer | None = None self._thread: threading.Thread | None = None def start(self): """Demarre le serveur dans un thread daemon.""" try: self._server = HTTPServer(("0.0.0.0", self._port), CaptureHandler) self._thread = threading.Thread( target=self._server.serve_forever, daemon=True ) self._thread.start() logger.info(f"Capture server demarre sur le port {self._port}") print(f"[CAPTURE] Serveur de capture demarre sur le port {self._port}") except Exception as e: logger.error(f"Impossible de demarrer le capture server : {e}") print(f"[CAPTURE] ERREUR demarrage : {e}") def stop(self): """Arrete le serveur proprement.""" if self._server: self._server.shutdown() logger.info("Capture server arrete")