Implement reconnect, idle reporting, and security hardening

src/lazy_enter/__main__.py

@@ -1,10 +1,20 @@
 """CLI 엔트리포인트."""
 
+import sys
+
 from lazy_enter.bridge import Bridge
+from lazy_enter.config import Config
 
 
 def main() -> None:
-    bridge = Bridge()
+    config = Config()
+    try:
+        config.validate_required_settings()
+    except ValueError as exc:
+        print(f"[LazyEnter] 설정 오류: {exc}", file=sys.stderr)
+        raise SystemExit(1) from exc
+
+    bridge = Bridge(config)
     bridge.run()
 
 

src/lazy_enter/bridge.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import logging
+import re
 import threading
 import time
 
@@ -23,18 +24,45 @@ class Bridge:
         self._output_thread: threading.Thread | None = None
         self._running = False
         self._channel: str = self.config.allowed_channel_id
+        self._last_input_at = time.monotonic()
+        self._last_output_at = time.monotonic()
+        self._input_idle_reported = False
+        self._output_idle_reported = False
 
         self.slack.on_message(self._handle_message)
         self.slack.on_command(self._handle_command)
 
     def _handle_message(self, text: str, channel: str) -> None:
         """Slack 메시지를 PTY 프로세스로 전달한다."""
+        if self._is_blocked_input(text):
+            self.slack.send_message(
+                channel, ":no_entry: 차단된 명령 패턴이 감지되었습니다."
+            )
+            return
+
         if self.pty and self.pty.is_alive:
             self.pty.send(text)
+            self._last_input_at = time.monotonic()
+            self._input_idle_reported = False
             logger.info("입력 전달: %s", text)
         else:
             self.slack.send_message(channel, ":warning: 실행 중인 세션이 없습니다.")
 
+    @staticmethod
+    def _is_blocked_input(text: str) -> bool:
+        """치명적 쉘 명령 패턴을 단순 차단한다."""
+        normalized = re.sub(r"\s+", " ", text.lower()).strip()
+        blocked_patterns = (
+            "rm -rf /",
+            "rm -rf /*",
+            "mkfs",
+            ":(){:|:&};:",
+            "shutdown -h",
+            "reboot",
+            "poweroff",
+        )
+        return any(pattern in normalized for pattern in blocked_patterns)
+
     def _handle_command(self, command: str, channel: str) -> None:
         """슬래시 커맨드를 처리한다."""
         if command == "start":
@@ -54,6 +82,10 @@ class Bridge:
         self.pty = PtyManager(self.config.default_shell)
         self.pty.start()
         self._running = True
+        self._last_input_at = time.monotonic()
+        self._last_output_at = time.monotonic()
+        self._input_idle_reported = False
+        self._output_idle_reported = False
         self._output_thread = threading.Thread(target=self._poll_output, daemon=True)
         self._output_thread.start()
 
@@ -71,9 +103,12 @@ class Bridge:
         """PTY 출력을 주기적으로 읽어 Slack으로 전송한다."""
         buffer = ""
         while self._running and self.pty and self.pty.is_alive:
+            now = time.monotonic()
             output = self.pty.read_output(timeout=self.config.pty_read_timeout)
             if output:
                 buffer += output
+                self._last_output_at = now
+                self._output_idle_reported = False
 
             if buffer:
                 # 메시지 길이 제한 적용
@@ -83,6 +118,33 @@ class Bridge:
                 self.slack.send_message(self._channel, f"```\n{message}\n```")
                 buffer = ""
 
+            output_idle = now - self._last_output_at
+            if (
+                self.config.output_idle_report_seconds > 0
+                and not self._output_idle_reported
+                and output_idle >= self.config.output_idle_report_seconds
+            ):
+                self.slack.send_message(
+                    self._channel,
+                    (
+                        f":hourglass_flowing_sand: 출력이 {int(output_idle)}초 동안 "
+                        "없습니다. 세션 상태를 확인해주세요."
+                    ),
+                )
+                self._output_idle_reported = True
+
+            input_idle = now - self._last_input_at
+            if (
+                self.config.input_idle_report_seconds > 0
+                and not self._input_idle_reported
+                and input_idle >= self.config.input_idle_report_seconds
+            ):
+                self.slack.send_message(
+                    self._channel,
+                    f":information_source: 입력이 {int(input_idle)}초 동안 없습니다.",
+                )
+                self._input_idle_reported = True
+
             time.sleep(self.config.output_buffer_interval)
 
         if not self._running:

src/lazy_enter/config.py

@@ -25,3 +25,26 @@ class Config:
     # Buffer
     output_buffer_interval: float = float(os.getenv("OUTPUT_BUFFER_INTERVAL", "2.0"))
     max_message_length: int = int(os.getenv("MAX_MESSAGE_LENGTH", "3000"))
+
+    # Status reporting / reconnect
+    reconnect_delay_seconds: float = float(os.getenv("RECONNECT_DELAY_SECONDS", "5.0"))
+    output_idle_report_seconds: int = int(
+        os.getenv("OUTPUT_IDLE_REPORT_SECONDS", "120")
+    )
+    input_idle_report_seconds: int = int(os.getenv("INPUT_IDLE_REPORT_SECONDS", "300"))
+
+    def validate_required_settings(self) -> None:
+        """필수 설정값 누락 여부를 검증한다."""
+        missing: list[str] = []
+        if not self.slack_bot_token:
+            missing.append("SLACK_BOT_TOKEN")
+        if not self.slack_app_token:
+            missing.append("SLACK_APP_TOKEN")
+        if not self.allowed_user_id:
+            missing.append("SLACK_ALLOWED_USER_ID")
+        if not self.allowed_channel_id:
+            missing.append("SLACK_ALLOWED_CHANNEL_ID")
+
+        if missing:
+            joined = ", ".join(missing)
+            raise ValueError(f"필수 환경 변수가 누락되었습니다: {joined}")

src/lazy_enter/slack_handler.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import logging
+import time
 
 from slack_bolt import App
 from slack_bolt.adapter.socket_mode import SocketModeHandler
@@ -19,7 +20,8 @@ class SlackHandler:
     def __init__(self, config: Config) -> None:
         self.config = config
         self.app = App(token=config.slack_bot_token)
-        self._handler = SocketModeHandler(self.app, config.slack_app_token)
+        self._handler: SocketModeHandler | None = None
+        self._stop_requested = False
         self._on_message_callback: callable | None = None
         self._on_command_callback: callable | None = None
 
@@ -98,11 +100,31 @@ class SlackHandler:
         )
 
     def start(self) -> None:
-        """Socket Mode 핸들러를 시작한다."""
+        """Socket Mode 핸들러를 시작한다. 연결이 끊기면 재연결한다."""
+        self._stop_requested = False
         logger.info("Slack Socket Mode 시작")
-        self._handler.start()
+
+        while not self._stop_requested:
+            try:
+                self._handler = SocketModeHandler(self.app, self.config.slack_app_token)
+                self._handler.start()
+            except Exception:
+                if self._stop_requested:
+                    break
+                logger.exception("Socket Mode 연결이 종료되었습니다.")
+
+            if self._stop_requested:
+                break
+
+            logger.warning(
+                "Socket Mode 재연결 시도 (%s초 후)",
+                self.config.reconnect_delay_seconds,
+            )
+            time.sleep(self.config.reconnect_delay_seconds)
 
     def stop(self) -> None:
         """Socket Mode 핸들러를 종료한다."""
         logger.info("Slack Socket Mode 종료")
-        self._handler.close()
+        self._stop_requested = True
+        if self._handler is not None:
+            self._handler.close()