Shared command shell adapter. Design goals: - absorb the useful command-cleaning behavior from legacy implementations - remove hardcoded xray fake responses and other test-only branches - avoid hardcoded prompt matching by using a command marker - support both local executi
| 104 | |
| 105 | |
| 106 | class InteractiveShell: |
| 107 | """ |
| 108 | Shared command shell adapter. |
| 109 | |
| 110 | Design goals: |
| 111 | - absorb the useful command-cleaning behavior from legacy implementations |
| 112 | - remove hardcoded xray fake responses and other test-only branches |
| 113 | - avoid hardcoded prompt matching by using a command marker |
| 114 | - support both local execution and delayed SSH connection |
| 115 | """ |
| 116 | |
| 117 | def __init__( |
| 118 | self, |
| 119 | config: RuntimeConfig | SSHConfig | None = None, |
| 120 | *, |
| 121 | scanner_provider: CommandToolProviderConfig | None = None, |
| 122 | ) -> None: |
| 123 | if isinstance(config, RuntimeConfig): |
| 124 | self.runtime = config |
| 125 | elif isinstance(config, SSHConfig): |
| 126 | self.runtime = RuntimeConfig(provider="ssh", ssh=config) |
| 127 | else: |
| 128 | self.runtime = RuntimeConfig() |
| 129 | self.scanner_provider = scanner_provider |
| 130 | self.client: Any | None = None |
| 131 | self.session: Any | None = None |
| 132 | |
| 133 | def _uses_ssh(self) -> bool: |
| 134 | return self.runtime.provider == "ssh" |
| 135 | |
| 136 | def _connect(self) -> None: |
| 137 | if not self._uses_ssh(): |
| 138 | return |
| 139 | if self.client is not None and self.session is not None: |
| 140 | return |
| 141 | |
| 142 | try: |
| 143 | import paramiko |
| 144 | except ImportError as exc: |
| 145 | raise RuntimeError( |
| 146 | "Paramiko is required for InteractiveShell. " |
| 147 | "Install project dependencies before using the terminal tool." |
| 148 | ) from exc |
| 149 | |
| 150 | client = paramiko.SSHClient() |
| 151 | client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) |
| 152 | client.connect( |
| 153 | self.runtime.ssh.host, |
| 154 | username=self.runtime.ssh.username, |
| 155 | password=self.runtime.ssh.password, |
| 156 | port=self.runtime.ssh.port, |
| 157 | timeout=self.runtime.ssh.timeout, |
| 158 | ) |
| 159 | self.client = client |
| 160 | self.session = client.invoke_shell() |
| 161 | self._drain_pending_output() |
| 162 | |
| 163 | def _drain_pending_output(self, idle_window: float = 0.2) -> None: |
no outgoing calls
no test coverage detected