diff --git a/README.md b/README.md index 75a28076..06fed1c2 100644 --- a/README.md +++ b/README.md @@ -13,6 +13,7 @@ ## Table of Contents +- [What's new (2026-06-19) — Unattended Reliability](#whats-new-2026-06-19--unattended-reliability) - [What's new (2026-06-19) — Popup Watchdog](#whats-new-2026-06-19--popup-watchdog) - [What's new (2026-06-19) — Native UI Control](#whats-new-2026-06-19--native-ui-control) - [What's new (2026-06-19)](#whats-new-2026-06-19) @@ -61,6 +62,14 @@ --- +## What's new (2026-06-19) — Unattended Reliability + +Three practitioner-pain fixes for unattended / login automation, all headless and full-stack. Full reference: [`docs/source/Eng/doc/new_features/v9_features_doc.rst`](docs/source/Eng/doc/new_features/v9_features_doc.rst). + +- **OTP / TOTP for 2FA** — `generate_totp` / `verify_totp` (`AC_otp_to_var`, `ac_generate_otp`): mint the current 6-digit code from a base32 secret to type into a login form (reuses the remote-desktop TOTP engine). +- **Native file dialogs** — `handle_file_dialog` (`AC_handle_file_dialog`): wait for the OS Open/Save/folder dialog, type the path, confirm — in one call, with an injectable driver. +- **Locked-session guard** — `ensure_interactive_session` / `is_session_locked` (`AC_assert_session_active`): fail clearly when the workstation is locked / disconnected instead of emitting phantom clicks. + ## What's new (2026-06-19) — Popup Watchdog The #1 cause of unattended-automation failure is an unexpected dialog the script never coded for (UAC, "session expiring", Windows Update, a modal). The popup watchdog runs a concurrent guard thread that watches for registered patterns and dismisses them independently of the main flow. Surfaced by the practitioner pain-point research as the top unattended failure cause; full stack (facade, `AC_*`, MCP, Script Builder), fully headless. Full reference: [`docs/source/Eng/doc/new_features/v8_features_doc.rst`](docs/source/Eng/doc/new_features/v8_features_doc.rst). diff --git a/README/README_zh-CN.md b/README/README_zh-CN.md index 43ae3234..73ba4e08 100644 --- a/README/README_zh-CN.md +++ b/README/README_zh-CN.md @@ -12,6 +12,7 @@ ## 目录 +- [本次更新 (2026-06-19) — 无人值守可靠性](#本次更新-2026-06-19--无人值守可靠性) - [本次更新 (2026-06-19) — 弹窗看门狗](#本次更新-2026-06-19--弹窗看门狗) - [本次更新 (2026-06-19) — 原生 UI 控制](#本次更新-2026-06-19--原生-ui-控制) - [本次更新 (2026-06-19)](#本次更新-2026-06-19) @@ -60,6 +61,14 @@ --- +## 本次更新 (2026-06-19) — 无人值守可靠性 + +三个无人值守/登录自动化的社区痛点修复,均 headless 且走完整五层。完整参考:[`docs/source/Eng/doc/new_features/v9_features_doc.rst`](../docs/source/Eng/doc/new_features/v9_features_doc.rst)。 + +- **2FA 的 OTP / TOTP** — `generate_totp` / `verify_totp`(`AC_otp_to_var`、`ac_generate_otp`):从 base32 secret 生成当下 6 位码,填进登录表单(重用远程桌面 TOTP 引擎)。 +- **原生文件对话框** — `handle_file_dialog`(`AC_handle_file_dialog`):等 OS 打开/保存/文件夹对话框、输入路径、确认,一次完成,driver 可注入。 +- **锁定会话守卫** — `ensure_interactive_session` / `is_session_locked`(`AC_assert_session_active`):工作站锁定/断开时清楚失败,而非发出幽灵点击。 + ## 本次更新 (2026-06-19) — 弹窗看门狗 无人值守自动化失败的第一大主因,是脚本没写到的意外对话框(UAC、"会话过期"、Windows Update、modal)。弹窗看门狗以并行守卫线程监看注册 pattern,独立于主流程把它们关掉。由社区痛点研究指出为无人值守头号失败主因;走完整五层(facade、`AC_*`、MCP、Script Builder),完全 headless。完整参考:[`docs/source/Eng/doc/new_features/v8_features_doc.rst`](../docs/source/Eng/doc/new_features/v8_features_doc.rst)。 diff --git a/README/README_zh-TW.md b/README/README_zh-TW.md index 7a6643c0..8f690127 100644 --- a/README/README_zh-TW.md +++ b/README/README_zh-TW.md @@ -12,6 +12,7 @@ ## 目錄 +- [本次更新 (2026-06-19) — 無人值守可靠性](#本次更新-2026-06-19--無人值守可靠性) - [本次更新 (2026-06-19) — 彈窗看門狗](#本次更新-2026-06-19--彈窗看門狗) - [本次更新 (2026-06-19) — 原生 UI 控制](#本次更新-2026-06-19--原生-ui-控制) - [本次更新 (2026-06-19)](#本次更新-2026-06-19) @@ -60,6 +61,14 @@ --- +## 本次更新 (2026-06-19) — 無人值守可靠性 + +三個無人值守/登入自動化的社群痛點修復,皆 headless 且走完整五層。完整參考:[`docs/source/Zh/doc/new_features/v9_features_doc.rst`](../docs/source/Zh/doc/new_features/v9_features_doc.rst)。 + +- **2FA 的 OTP / TOTP** — `generate_totp` / `verify_totp`(`AC_otp_to_var`、`ac_generate_otp`):從 base32 secret 產生當下 6 碼,填進登入表單(重用遠端桌面 TOTP 引擎)。 +- **原生檔案對話框** — `handle_file_dialog`(`AC_handle_file_dialog`):等 OS 開啟/儲存/資料夾對話框、輸入路徑、確認,一次完成,driver 可注入。 +- **鎖定工作階段守衛** — `ensure_interactive_session` / `is_session_locked`(`AC_assert_session_active`):工作站鎖定/斷線時清楚失敗,而非送出幽靈點擊。 + ## 本次更新 (2026-06-19) — 彈窗看門狗 無人值守自動化失敗的第一大主因,是腳本沒寫到的未預期對話框(UAC、「工作階段過期」、Windows Update、modal)。彈窗看門狗以並行守衛執行緒監看註冊 pattern,獨立於主流程把它們關掉。由社群痛點研究指出為無人值守頭號失敗主因;走完整五層(facade、`AC_*`、MCP、Script Builder),完全 headless。完整參考:[`docs/source/Zh/doc/new_features/v8_features_doc.rst`](../docs/source/Zh/doc/new_features/v8_features_doc.rst)。 diff --git a/docs/source/Eng/doc/new_features/v9_features_doc.rst b/docs/source/Eng/doc/new_features/v9_features_doc.rst new file mode 100644 index 00000000..121a70b0 --- /dev/null +++ b/docs/source/Eng/doc/new_features/v9_features_doc.rst @@ -0,0 +1,70 @@ +=================================================== +New Features (2026-06-19) — Unattended Reliability +=================================================== + +Three practitioner-pain fixes for unattended and login automation: mint +2FA codes, drive native file dialogs, and refuse to act on a locked +screen. Each ships through the full stack (facade, ``AC_*`` executor +commands, MCP tools, Script Builder) and is fully headless — external +steps are deterministic or injectable, so they unit-test without 2FA, a +real dialog, or a locked session. + +.. contents:: + :local: + :depth: 2 + + +OTP / TOTP for 2FA logins +========================= + +2FA blocks automated logins. Store the base32 secret (ideally in the +secrets store) and mint the current code mid-flow:: + + from je_auto_control import generate_totp, verify_totp + + code = generate_totp(secret) # 6-digit TOTP for "now" + type_text(code) + +``AC_otp_to_var`` writes the code into a flow variable for the next step:: + + ["AC_otp_to_var", {"secret": "JBSWY3DPEHPK3PXP", "var": "otp"}] + ["AC_type_keyboard", {"keycode": "${otp}"}] + +Reuses the TOTP engine that backs remote-desktop auth. Executor command: +``AC_otp_to_var``; MCP tool: ``ac_generate_otp``. + + +Native file dialogs +=================== + +Recorders don't capture the OS file Open/Save/folder dialog, so everyone +hand-rolls "type the path + Enter". ``handle_file_dialog`` does it in one +call:: + + from je_auto_control import handle_file_dialog + + handle_file_dialog("C:/reports/out.csv", action="save") + +``action`` is ``open`` / ``save`` / ``folder`` (picking a default dialog +title) or pass an explicit ``window_title``; it waits for the dialog, +types the path, and presses ``confirm_key`` (default Enter). The +window-wait / type / confirm steps go through an injectable +:class:`FileDialogDriver`. Executor command: ``AC_handle_file_dialog``. + + +Locked-session guard +=================== + +Unattended runs silently fail when the workstation is locked or the RDP +session is disconnected — input no-ops or throws. Check first:: + + from je_auto_control import ensure_interactive_session, is_session_locked + + ensure_interactive_session() # raises if locked + if is_session_locked(): + ... + +On Windows the probe opens the input desktop (which fails when locked); +other platforms report "not locked" unless a custom probe is supplied. +Executor command: ``AC_assert_session_active`` — put it at the top of an +unattended script so it fails clearly instead of emitting phantom clicks. diff --git a/docs/source/Eng/eng_index.rst b/docs/source/Eng/eng_index.rst index 8df2c67e..2c63bef7 100644 --- a/docs/source/Eng/eng_index.rst +++ b/docs/source/Eng/eng_index.rst @@ -31,6 +31,7 @@ Comprehensive guides for all AutoControl features. doc/new_features/v6_features_doc doc/new_features/v7_features_doc doc/new_features/v8_features_doc + doc/new_features/v9_features_doc doc/ocr_backends/ocr_backends_doc doc/observability/observability_doc doc/operations_layer/operations_layer_doc diff --git a/docs/source/Zh/doc/new_features/v9_features_doc.rst b/docs/source/Zh/doc/new_features/v9_features_doc.rst new file mode 100644 index 00000000..4d3b16df --- /dev/null +++ b/docs/source/Zh/doc/new_features/v9_features_doc.rst @@ -0,0 +1,66 @@ +======================================== +新功能 (2026-06-19) — 無人值守可靠性 +======================================== + +三個針對無人值守與登入自動化的社群痛點修復:產生 2FA 驗證碼、操作原生 +檔案對話框、以及拒絕在鎖定畫面上動作。每項都走完整五層(facade、 +``AC_*`` 執行器指令、MCP 工具、Script Builder),且完全 headless——外部 +步驟皆為決定性或可注入,因此不需 2FA、真實對話框或鎖定工作階段即可單元 +測試。 + +.. contents:: + :local: + :depth: 2 + + +2FA 登入的 OTP / TOTP +===================== + +2FA 會擋住自動登入。把 base32 secret 存好(最好放進 secrets store),在 +流程中即時產生當前驗證碼:: + + from je_auto_control import generate_totp, verify_totp + + code = generate_totp(secret) # 當下的 6 碼 TOTP + type_text(code) + +``AC_otp_to_var`` 會把驗證碼寫進流程變數供下一步使用:: + + ["AC_otp_to_var", {"secret": "JBSWY3DPEHPK3PXP", "var": "otp"}] + ["AC_type_keyboard", {"keycode": "${otp}"}] + +重用支撐遠端桌面驗證的 TOTP 引擎。執行器指令:``AC_otp_to_var``; +MCP 工具:``ac_generate_otp``。 + + +原生檔案對話框 +============== + +錄製器抓不到 OS 的檔案開啟/儲存/資料夾對話框,大家只好手刻「輸入路徑 + +Enter」。``handle_file_dialog`` 一次搞定:: + + from je_auto_control import handle_file_dialog + + handle_file_dialog("C:/reports/out.csv", action="save") + +``action`` 為 ``open`` / ``save`` / ``folder``(自動選預設對話框標題), +或傳明確的 ``window_title``;它會等對話框、輸入路徑、按 ``confirm_key`` +(預設 Enter)。等視窗/輸入/確認三步透過可注入的 :class:`FileDialogDriver`。 +執行器指令:``AC_handle_file_dialog``。 + + +鎖定工作階段守衛 +================ + +無人值守在工作站鎖定或 RDP 斷線時會默默失敗——輸入會 no-op 或拋例外。 +先檢查:: + + from je_auto_control import ensure_interactive_session, is_session_locked + + ensure_interactive_session() # 鎖定時拋例外 + if is_session_locked(): + ... + +Windows 上以開啟 input desktop 偵測(鎖定時會失敗);其他平台除非提供自訂 +probe,否則回報「未鎖定」。執行器指令:``AC_assert_session_active``——放在 +無人值守腳本最前面,讓它清楚地失敗,而非送出幽靈點擊。 diff --git a/docs/source/Zh/zh_index.rst b/docs/source/Zh/zh_index.rst index b9250a7c..d6fe8a69 100644 --- a/docs/source/Zh/zh_index.rst +++ b/docs/source/Zh/zh_index.rst @@ -31,6 +31,7 @@ AutoControl 所有功能的完整使用指南。 doc/new_features/v6_features_doc doc/new_features/v7_features_doc doc/new_features/v8_features_doc + doc/new_features/v9_features_doc doc/ocr_backends/ocr_backends_doc doc/observability/observability_doc doc/operations_layer/operations_layer_doc diff --git a/je_auto_control/__init__.py b/je_auto_control/__init__.py index 35136b90..77ea570c 100644 --- a/je_auto_control/__init__.py +++ b/je_auto_control/__init__.py @@ -97,6 +97,18 @@ from je_auto_control.utils.hotkey.hotkey_daemon import ( HotkeyBinding, HotkeyDaemon, default_hotkey_daemon, ) +# OTP/TOTP for automated 2FA logins +from je_auto_control.utils.otp import ( + TOTPError, generate_secret, generate_totp, verify_totp, +) +# Native file Open/Save/folder dialog helper +from je_auto_control.utils.file_dialog import ( + FileDialogDriver, handle_file_dialog, +) +# Locked / non-interactive session guard +from je_auto_control.utils.session_guard import ( + ensure_interactive_session, is_session_locked, +) # Background popup/interrupt watchdog (unattended automation) from je_auto_control.utils.watchdog import ( PopupWatchdog, WatchdogRule, default_popup_watchdog, @@ -494,6 +506,9 @@ def start_autocontrol_gui(*args, **kwargs): # Hotkey daemon "HotkeyDaemon", "HotkeyBinding", "default_hotkey_daemon", "PopupWatchdog", "WatchdogRule", "default_popup_watchdog", + "generate_totp", "verify_totp", "generate_secret", "TOTPError", + "handle_file_dialog", "FileDialogDriver", + "ensure_interactive_session", "is_session_locked", # MCP server "AuditLogger", "HttpMCPServer", "MCPContent", "MCPPrompt", "MCPPromptArgument", "MCPResource", "MCPServer", "MCPTool", diff --git a/je_auto_control/gui/script_builder/command_schema.py b/je_auto_control/gui/script_builder/command_schema.py index 279e3a4f..54c1fe9e 100644 --- a/je_auto_control/gui/script_builder/command_schema.py +++ b/je_auto_control/gui/script_builder/command_schema.py @@ -623,6 +623,35 @@ def _add_misc_specs(specs: List[CommandSpec]) -> None: "AC_watchdog_stop", "Flow", "Watchdog: Stop")) specs.append(CommandSpec( "AC_watchdog_list", "Flow", "Watchdog: List Rules / Hits")) + specs.append(CommandSpec( + "AC_otp_to_var", "Flow", "OTP (TOTP) into Variable", + fields=( + FieldSpec("secret", FieldType.STRING), + FieldSpec("var", FieldType.STRING, default="otp"), + FieldSpec("digits", FieldType.INT, optional=True, default=6), + FieldSpec("step", FieldType.INT, optional=True, default=30), + ), + description="Generate a TOTP 2FA code from a base32 secret.", + )) + specs.append(CommandSpec( + "AC_handle_file_dialog", "Native UI", "Handle File Dialog", + fields=( + FieldSpec("path", FieldType.STRING), + FieldSpec("action", FieldType.ENUM, + choices=("open", "save", "folder"), + optional=True, default="open"), + FieldSpec("window_title", FieldType.STRING, optional=True), + FieldSpec("timeout_s", FieldType.FLOAT, optional=True, + default=10.0), + FieldSpec("confirm_key", FieldType.STRING, optional=True, + default="enter"), + ), + description="Wait for a native file dialog, type a path, confirm.", + )) + specs.append(CommandSpec( + "AC_assert_session_active", "Flow", "Assert Session Active", + description="Fail if the session is locked / non-interactive.", + )) specs.append(CommandSpec( "AC_shell_command", "Shell", "Shell Command", fields=(FieldSpec("shell_command", FieldType.STRING),), diff --git a/je_auto_control/utils/executor/action_executor.py b/je_auto_control/utils/executor/action_executor.py index e71d2151..e2de074c 100644 --- a/je_auto_control/utils/executor/action_executor.py +++ b/je_auto_control/utils/executor/action_executor.py @@ -2289,6 +2289,23 @@ def _watchdog_list() -> Dict[str, Any]: return {"running": w.running, "rules": w.rule_names(), "hits": w.hits} +def _handle_file_dialog(path: str, action: str = "open", + window_title: Optional[str] = None, + timeout_s: float = 10.0, + confirm_key: str = "enter") -> Dict[str, Any]: + """Adapter: wait for a native file dialog, type the path, confirm.""" + from je_auto_control.utils.file_dialog import handle_file_dialog + return handle_file_dialog(path, action=action, window_title=window_title, + timeout_s=float(timeout_s), + confirm_key=confirm_key) + + +def _assert_session_active() -> Dict[str, Any]: + """Adapter: raise unless the session is interactive (not locked).""" + from je_auto_control.utils.session_guard import ensure_interactive_session + return {"interactive": ensure_interactive_session()} + + class Executor: """ Executor @@ -2445,6 +2462,8 @@ def __init__(self): "AC_watchdog_start": _watchdog_start, "AC_watchdog_stop": _watchdog_stop, "AC_watchdog_list": _watchdog_list, + "AC_handle_file_dialog": _handle_file_dialog, + "AC_assert_session_active": _assert_session_active, "AC_a11y_record_start": _a11y_record_start, "AC_a11y_record_stop": _a11y_record_stop, "AC_a11y_record_events": _a11y_record_events, diff --git a/je_auto_control/utils/executor/flow_control.py b/je_auto_control/utils/executor/flow_control.py index fc13ebaa..4c8c6a58 100644 --- a/je_auto_control/utils/executor/flow_control.py +++ b/je_auto_control/utils/executor/flow_control.py @@ -440,6 +440,16 @@ def exec_pdf_to_var(executor: Any, args: Mapping[str, Any]) -> Dict[str, Any]: return {"var": var_name, "length": len(text)} +def exec_otp_to_var(executor: Any, args: Mapping[str, Any]) -> Dict[str, Any]: + """Generate a TOTP code from a base32 secret into a flow variable (2FA).""" + from je_auto_control.utils.otp import generate_totp + code = generate_totp(args["secret"], step=int(args.get("step", 30)), + digits=int(args.get("digits", 6))) + var_name = args.get("var", "otp") + executor.variables.set(var_name, code) + return {"var": var_name} + + def exec_sql_to_var(executor: Any, args: Mapping[str, Any]) -> Dict[str, Any]: """Run a read-only SQLite query and store its result in a flow variable.""" from je_auto_control.utils.sql.sql_query import query_sqlite @@ -670,6 +680,7 @@ def exec_call_macro(executor: Any, args: Mapping[str, Any]) -> Any: "AC_shell_to_var": exec_shell_to_var, "AC_read_file_to_var": exec_read_file_to_var, "AC_pdf_to_var": exec_pdf_to_var, + "AC_otp_to_var": exec_otp_to_var, "AC_sql_to_var": exec_sql_to_var, "AC_assert_db": exec_assert_db, "AC_http_to_var": exec_http_to_var, diff --git a/je_auto_control/utils/file_dialog/__init__.py b/je_auto_control/utils/file_dialog/__init__.py new file mode 100644 index 00000000..f60bccc5 --- /dev/null +++ b/je_auto_control/utils/file_dialog/__init__.py @@ -0,0 +1,6 @@ +"""Drive native file Open/Save/folder-picker dialogs.""" +from je_auto_control.utils.file_dialog.file_dialog import ( + FileDialogDriver, handle_file_dialog, +) + +__all__ = ["FileDialogDriver", "handle_file_dialog"] diff --git a/je_auto_control/utils/file_dialog/file_dialog.py b/je_auto_control/utils/file_dialog/file_dialog.py new file mode 100644 index 00000000..1b807f56 --- /dev/null +++ b/je_auto_control/utils/file_dialog/file_dialog.py @@ -0,0 +1,54 @@ +"""Handle native file Open / Save-As / folder-picker dialogs. + +A universal RPA pain: recorders don't capture the OS file dialog, so +everyone hand-rolls "type the full path + Enter". This waits for the +native dialog window, types the path into it, and confirms — in one call. + +The window-wait / type / confirm steps go through an injectable +:class:`FileDialogDriver` so the orchestration is unit-tested without a +real dialog; the default driver uses the window + keyboard wrappers. +Imports no ``PySide6``. +""" +from typing import Dict, Optional + +_DEFAULT_TITLES = {"open": "Open", "save": "Save As", "folder": "Select Folder"} + + +class FileDialogDriver: + """Pluggable window-wait / type / confirm steps for a file dialog.""" + + def wait_window(self, title: str, timeout_s: float) -> bool: + from je_auto_control.wrapper.auto_control_window import wait_for_window + try: + wait_for_window(title, timeout=float(timeout_s)) + return True + except (OSError, RuntimeError, ValueError): + return False + + def type_path(self, path: str) -> None: + from je_auto_control.wrapper.auto_control_keyboard import write + write(str(path)) + + def confirm(self, key: str) -> None: + from je_auto_control.wrapper.auto_control_keyboard import type_keyboard + type_keyboard(str(key)) + + +def handle_file_dialog(path: str, *, action: str = "open", + window_title: Optional[str] = None, + timeout_s: float = 10.0, confirm_key: str = "enter", + driver: Optional[FileDialogDriver] = None, + ) -> Dict[str, object]: + """Wait for a native file dialog, type ``path``, and confirm. + + :param action: ``open`` / ``save`` / ``folder`` — picks a default + dialog title when ``window_title`` is not given. + :returns: ``{"handled": bool, "title": str}``. + """ + title = window_title or _DEFAULT_TITLES.get(action, _DEFAULT_TITLES["open"]) + drv = driver or FileDialogDriver() + if not drv.wait_window(title, float(timeout_s)): + return {"handled": False, "title": title} + drv.type_path(str(path)) + drv.confirm(str(confirm_key)) + return {"handled": True, "title": title} diff --git a/je_auto_control/utils/mcp_server/tools/_factories.py b/je_auto_control/utils/mcp_server/tools/_factories.py index e07d3fb9..d1ddd73e 100644 --- a/je_auto_control/utils/mcp_server/tools/_factories.py +++ b/je_auto_control/utils/mcp_server/tools/_factories.py @@ -1669,6 +1669,50 @@ def process_and_shell_tools() -> List[MCPTool]: ] +def unattended_tools() -> List[MCPTool]: + return [ + MCPTool( + name="ac_generate_otp", + description=("Generate the current TOTP code from a base32 secret " + "for automated 2FA logins. step/digits default to " + "30/6. Returns the numeric code string."), + input_schema=schema({ + "secret": {"type": "string"}, + "step": {"type": "integer"}, + "digits": {"type": "integer"}, + }, required=["secret"]), + handler=h.generate_otp, + annotations=READ_ONLY, + ), + MCPTool( + name="ac_handle_file_dialog", + description=("Wait for a native file dialog (action=open|save|" + "folder, or a custom window_title), type 'path' into " + "it, and confirm (default Enter). Returns " + "{handled, title}."), + input_schema=schema({ + "path": {"type": "string"}, + "action": {"type": "string"}, + "window_title": {"type": "string"}, + "timeout_s": {"type": "number"}, + "confirm_key": {"type": "string"}, + }, required=["path"]), + handler=h.handle_file_dialog, + annotations=DESTRUCTIVE, + ), + MCPTool( + name="ac_assert_session_active", + description=("Raise when the interactive session is locked / " + "disconnected (so an unattended run fails clearly " + "instead of emitting phantom input). Returns " + "{interactive: true} when OK."), + input_schema=schema({}), + handler=h.assert_session_active, + annotations=READ_ONLY, + ), + ] + + def watchdog_tools() -> List[MCPTool]: return [ MCPTool( @@ -2654,6 +2698,7 @@ def media_assert_tools() -> List[MCPTool]: computer_use_tools, dag_tools, presence_tools, chatops_tools, redaction_tools, android_widget_tools, ios_tools, webrunner_tools, scheduler_tools, trigger_tools, hotkey_tools, watchdog_tools, + unattended_tools, screen_record_tools, process_and_shell_tools, remote_desktop_tools, gamepad_tools, usb_passthrough_tools, assertion_tools, data_source_tools, diff --git a/je_auto_control/utils/mcp_server/tools/_handlers.py b/je_auto_control/utils/mcp_server/tools/_handlers.py index 0eff75fa..77b1ddcd 100644 --- a/je_auto_control/utils/mcp_server/tools/_handlers.py +++ b/je_auto_control/utils/mcp_server/tools/_handlers.py @@ -776,6 +776,23 @@ def watchdog_list(): return {"running": w.running, "rules": w.rule_names(), "hits": w.hits} +def generate_otp(secret, step=30, digits=6): + from je_auto_control.utils.otp import generate_totp + return generate_totp(secret, step=int(step), digits=int(digits)) + + +def handle_file_dialog(path, action="open", window_title=None, + timeout_s=10.0, confirm_key="enter"): + from je_auto_control.utils.file_dialog import handle_file_dialog as _h + return _h(path, action=action, window_title=window_title, + timeout_s=float(timeout_s), confirm_key=confirm_key) + + +def assert_session_active(): + from je_auto_control.utils.session_guard import ensure_interactive_session + return {"interactive": ensure_interactive_session()} + + def vlm_locate(description: str, screen_region: Optional[List[int]] = None, model: Optional[str] = None) -> Optional[List[int]]: diff --git a/je_auto_control/utils/otp/__init__.py b/je_auto_control/utils/otp/__init__.py new file mode 100644 index 00000000..098cf7f7 --- /dev/null +++ b/je_auto_control/utils/otp/__init__.py @@ -0,0 +1,6 @@ +"""One-time-password (TOTP) generation for automated 2FA logins.""" +from je_auto_control.utils.otp.otp import ( + TOTPError, generate_secret, generate_totp, verify_totp, +) + +__all__ = ["TOTPError", "generate_secret", "generate_totp", "verify_totp"] diff --git a/je_auto_control/utils/otp/otp.py b/je_auto_control/utils/otp/otp.py new file mode 100644 index 00000000..ccf959a5 --- /dev/null +++ b/je_auto_control/utils/otp/otp.py @@ -0,0 +1,31 @@ +"""Generate TOTP codes for automated 2FA logins. + +2FA blocks unattended logins; teams bolt on pyotp/Twilio glue to mint a +code mid-flow. This exposes the TOTP engine that already backs +remote-desktop auth as a general flow utility: store the base32 secret +(ideally in the secrets store) and mint the current code to type into a +login form. Imports no ``PySide6``. +""" +from typing import Optional + +from je_auto_control.utils.remote_desktop.totp import ( + TOTPError, generate_code, generate_secret, verify_code, +) + +__all__ = ["TOTPError", "generate_secret", "generate_totp", "verify_totp"] + + +def generate_totp(secret: str, *, at: Optional[float] = None, + step: int = 30, digits: int = 6) -> str: + """Return the current TOTP code for a base32 ``secret``. + + ``at`` spoofs the time (seconds since epoch) for deterministic tests. + """ + return generate_code(secret, at=at, step=int(step), digits=int(digits)) + + +def verify_totp(secret: str, code: str, *, at: Optional[float] = None, + step: int = 30, digits: int = 6, window: int = 1) -> bool: + """Return True if ``code`` is valid for ``secret`` (± ``window`` steps).""" + return verify_code(secret, code, at=at, step=int(step), + digits=int(digits), window=int(window)) diff --git a/je_auto_control/utils/remote_desktop/totp.py b/je_auto_control/utils/remote_desktop/totp.py index 26c21dfe..07dd0c6d 100644 --- a/je_auto_control/utils/remote_desktop/totp.py +++ b/je_auto_control/utils/remote_desktop/totp.py @@ -92,9 +92,10 @@ def verify_code(secret: str, code: str, *, now = time.time() if at is None else at base_counter = int(now) // step for delta in range(-window, window + 1): - expected = _code_for_counter( - decoded, base_counter + delta, digits=digits, - ) + counter = base_counter + delta + if counter < 0: # near the epoch the lower window can go negative + continue + expected = _code_for_counter(decoded, counter, digits=digits) if hmac.compare_digest(expected, cleaned): return True return False diff --git a/je_auto_control/utils/session_guard/__init__.py b/je_auto_control/utils/session_guard/__init__.py new file mode 100644 index 00000000..1ec91587 --- /dev/null +++ b/je_auto_control/utils/session_guard/__init__.py @@ -0,0 +1,6 @@ +"""Detect a locked / non-interactive session before driving input.""" +from je_auto_control.utils.session_guard.session_guard import ( + ensure_interactive_session, is_session_locked, +) + +__all__ = ["ensure_interactive_session", "is_session_locked"] diff --git a/je_auto_control/utils/session_guard/session_guard.py b/je_auto_control/utils/session_guard/session_guard.py new file mode 100644 index 00000000..202bedb8 --- /dev/null +++ b/je_auto_control/utils/session_guard/session_guard.py @@ -0,0 +1,56 @@ +"""Guard against driving input into a locked / non-interactive session. + +Unattended runs silently fail when the workstation is locked or the RDP +session is disconnected: ``SetCursorPos`` / clicks no-op or throw, and the +script produces phantom input. Check first and fail with a clear error +instead. On Windows the probe opens the input desktop (which fails when +the station is locked); other platforms report "not locked" unless a +custom probe is supplied. The probe is injectable for tests. Imports no +``PySide6``. +""" +import sys +from typing import Callable, Optional + +from je_auto_control.utils.exception.exceptions import AutoControlException + +LockProbe = Callable[[], bool] + + +def _windows_locked() -> bool: + """True when the Windows input desktop can't be opened (station locked).""" + import ctypes + _DESKTOP_READOBJECTS = 0x0001 + user32 = ctypes.windll.user32 # nosec B607 # reason: fixed system DLL + handle = user32.OpenInputDesktop(0, False, _DESKTOP_READOBJECTS) + if not handle: + return True + user32.CloseDesktop(handle) + return False + + +def _default_probe() -> bool: + """Best-effort lock detection; only implemented on Windows.""" + if sys.platform in ("win32", "cygwin", "msys"): + try: + return _windows_locked() + except (OSError, AttributeError): + return False + return False + + +def is_session_locked(probe: Optional[LockProbe] = None) -> bool: + """Return True when the interactive session appears locked / unavailable.""" + return bool((probe or _default_probe)()) + + +def ensure_interactive_session(probe: Optional[LockProbe] = None) -> bool: + """Raise :class:`AutoControlException` when the session is locked. + + Returns True when interactive, so it can gate an unattended flow before + it emits phantom clicks into a locked screen. + """ + if is_session_locked(probe): + raise AutoControlException( + "session is locked or non-interactive; input would be lost", + ) + return True diff --git a/test/unit_test/headless/test_unattended_reliability.py b/test/unit_test/headless/test_unattended_reliability.py new file mode 100644 index 00000000..d7677ef5 --- /dev/null +++ b/test/unit_test/headless/test_unattended_reliability.py @@ -0,0 +1,95 @@ +"""Headless tests for the unattended-reliability features: +OTP/TOTP generation, native file-dialog handling, and the session guard. +All external steps are deterministic or injected — no real GUI/2FA.""" +import pytest + +import je_auto_control as ac +from je_auto_control.utils.exception.exceptions import AutoControlException + + +# --- OTP / TOTP ------------------------------------------------------------ + +def test_generate_totp_deterministic_and_verifies(): + secret = ac.generate_secret() + code = ac.generate_totp(secret, at=0) + assert code.isdigit() and len(code) == 6 + assert ac.verify_totp(secret, code, at=0) is True + # same 30-second step -> same code + assert ac.generate_totp(secret, at=0) == ac.generate_totp(secret, at=15) + # a code from a far-away step does not verify (near-epoch counters safe) + assert ac.verify_totp(secret, ac.generate_totp(secret, at=99999), + at=0) is False + + +def test_otp_to_var_command(): + from je_auto_control.utils.executor.action_executor import Executor + from je_auto_control.utils.executor.flow_control import exec_otp_to_var + executor = Executor() + result = exec_otp_to_var( + executor, {"secret": ac.generate_secret(), "var": "code"}) + assert result["var"] == "code" + assert executor.variables.get_value("code").isdigit() + + +# --- native file dialog ---------------------------------------------------- + +class _FakeDriver: + def __init__(self, found=True): + self.found = found + self.calls = {} + + def wait_window(self, title, timeout_s): + self.calls["wait"] = (title, timeout_s) + return self.found + + def type_path(self, path): + self.calls["typed"] = path + + def confirm(self, key): + self.calls["confirm"] = key + + +def test_handle_file_dialog_types_and_confirms(): + from je_auto_control.utils.file_dialog import handle_file_dialog + drv = _FakeDriver() + result = handle_file_dialog("C:/out.txt", action="save", driver=drv) + assert result == {"handled": True, "title": "Save As"} + assert drv.calls["typed"] == "C:/out.txt" + assert drv.calls["confirm"] == "enter" + + +def test_handle_file_dialog_missing_dialog(): + from je_auto_control.utils.file_dialog import handle_file_dialog + drv = _FakeDriver(found=False) + result = handle_file_dialog("x", driver=drv) + assert result["handled"] is False + assert "typed" not in drv.calls + + +# --- session guard --------------------------------------------------------- + +def test_session_locked_raises(): + assert ac.is_session_locked(probe=lambda: True) is True + with pytest.raises(AutoControlException): + ac.ensure_interactive_session(probe=lambda: True) + + +def test_session_active_passes(): + assert ac.is_session_locked(probe=lambda: False) is False + assert ac.ensure_interactive_session(probe=lambda: False) is True + + +# --- wiring ---------------------------------------------------------------- + +def test_executor_and_mcp_and_builder_wiring(): + known = ac.executor.known_commands() + assert {"AC_otp_to_var", "AC_handle_file_dialog", + "AC_assert_session_active"} <= known + from je_auto_control.utils.mcp_server.tools import build_default_tool_registry + names = {t.name for t in build_default_tool_registry()} + assert {"ac_generate_otp", "ac_handle_file_dialog", + "ac_assert_session_active"} <= names + from je_auto_control.gui.script_builder.command_schema import _build_specs + cmds = {s.command for s in _build_specs()} + assert {"AC_otp_to_var", "AC_handle_file_dialog", + "AC_assert_session_active"} <= cmds