diff --git a/README.md b/README.md index 03703ba..b51ed07 100644 --- a/README.md +++ b/README.md @@ -122,9 +122,10 @@ sudo apt install -y python3-gi python3-gi-cairo gir1.2-gtk-3.0 gir1.2-webkit2-4. - **Dark cybersecurity theme** with a polished, modern interface - **Real-time streaming** — AI responses stream in via Server-Sent Events (SSE) - **All modes accessible** — Chat, Agent, Plan, CVE Lookup, OSINT, Topology, Compliance panels +- **Collapsible sidebar** — Minimize navigation to an icon rail with the preference remembered locally - **Session management** — Browse, restore, and delete saved sessions from the sidebar - **Provider switching** — Change AI provider and model on the fly from the settings panel -- **Agent control panel** — Start assessments, step through actions, view findings live +- **Agent control panel** — Start assessments that auto-run steps, stop them on demand, and view findings live - **CVE search panel** — Search by keyword, CVE ID, or browse exploits with severity filters - **OSINT panel** — Run subdomain enumeration, DNS lookups, WHOIS, and tech fingerprinting - **Topology visualization** — Paste nmap/masscan output and see an interactive D3.js network graph diff --git a/docs/superpowers/plans/2026-06-21-deepseek-multi-key-failover.md b/docs/superpowers/plans/2026-06-21-deepseek-multi-key-failover.md new file mode 100644 index 0000000..4b36023 --- /dev/null +++ b/docs/superpowers/plans/2026-06-21-deepseek-multi-key-failover.md @@ -0,0 +1,1100 @@ +# DeepSeek Multi-Key Failover Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Let users store a pool of DeepSeek API keys so HackBot automatically rotates to the next working key and retries the request when the active key is dead or out of credit. + +**Architecture:** A key pool (`AIConfig.api_keys`) persists in `config.yaml` with the active key mirrored into the existing `api_key` field for backward compatibility. `AIEngine` classifies failures and, only for the `deepseek` provider, rotates to the next untried key and retries — transparently for blocking calls, and (for streaming) only when no token has been emitted yet. REPL commands manage the pool. + +**Tech Stack:** Python 3.9+, OpenAI SDK (provider-agnostic client), Click + Rich CLI, pytest with `unittest.mock`. + +**Spec:** `docs/superpowers/specs/2026-06-21-deepseek-multi-key-failover-design.md` + +--- + +## File Structure + +- `hackbot/config.py` — add `api_keys` field, `reconcile_keys()`, pool helpers (`add_key`/`remove_key`/`set_active_key`), `mask_key()`. Persist via `save_config`. +- `hackbot/core/engine.py` — `KeyPoolExhaustedError`, `_key_failure_reason()` classifier, rotation state + `_maybe_rotate()`, failover loops in `_blocking_chat`/`_stream_chat`, `on_notice` hook. +- `hackbot/cli.py` — `_rebuild_engine()` helper, `_set_key()` subcommand routing, `_key_add`/`_key_remove`/`_list_keys`, `/keys` dispatch entry. +- `hackbot/ui/__init__.py` — `/help` text additions. +- `tests/test_config.py`, `tests/test_engine.py`, `tests/test_keypool_cli.py` — coverage. + +--- + +## Task 1: Config — `api_keys` field, persistence & reconciliation + +**Files:** +- Modify: `hackbot/config.py` +- Test: `tests/test_config.py` + +- [ ] **Step 1: Write the failing tests** + +Add to `tests/test_config.py` (extend the import from `hackbot.config` to include `reconcile_keys`, `save_config`): + +```python +from hackbot.config import reconcile_keys, save_config + + +def test_reconcile_seeds_pool_from_single_key(): + ai = {"api_key": "sk-a", "api_keys": []} + reconcile_keys(ai) + assert ai["api_keys"] == ["sk-a"] + assert ai["api_key"] == "sk-a" + + +def test_reconcile_active_is_pool_first(): + ai = {"api_key": "", "api_keys": ["sk-a", "sk-b"]} + reconcile_keys(ai) + assert ai["api_key"] == "sk-a" + + +def test_reconcile_env_key_prepended_and_deduped(): + ai = {"api_key": "sk-env", "api_keys": ["sk-a", "sk-env", "sk-b"]} + reconcile_keys(ai) + assert ai["api_keys"] == ["sk-env", "sk-a", "sk-b"] + assert ai["api_key"] == "sk-env" + + +def test_reconcile_empty_pool(): + ai = {"api_key": "", "api_keys": []} + reconcile_keys(ai) + assert ai["api_keys"] == [] + assert ai["api_key"] == "" + + +def test_api_keys_round_trip(tmp_path, monkeypatch): + import hackbot.config as cfgmod + monkeypatch.setattr(cfgmod, "CONFIG_FILE", tmp_path / "config.yaml") + for var in ("DEEPSEEK_API_KEY", "OPENAI_API_KEY", "HACKBOT_API_KEY", "ANTHROPIC_API_KEY"): + monkeypatch.delenv(var, raising=False) + + cfg = cfgmod.HackBotConfig() + cfg.ai.provider = "deepseek" + cfg.ai.api_keys = ["sk-a", "sk-b"] + cfg.ai.api_key = "sk-a" + cfgmod.save_config(cfg) + + loaded = cfgmod.load_config() + assert loaded.ai.api_keys == ["sk-a", "sk-b"] + assert loaded.ai.api_key == "sk-a" +``` + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `pytest tests/test_config.py -k "reconcile or round_trip" -v` +Expected: FAIL — `ImportError: cannot import name 'reconcile_keys'` (and `AIConfig` has no `api_keys`). + +- [ ] **Step 3: Implement** + +In `hackbot/config.py`: + +3a. Add `api_keys` to the default config — inside `DEFAULT_CONFIG["ai"]`, after `"base_url": "",`: + +```python + "api_keys": [], +``` + +3b. Add the field to `AIConfig` (after `api_key: str = ""`): + +```python + api_keys: List[str] = field(default_factory=list) +``` + +3c. Persist it — in `save_config`, inside the `"ai": { ... }` dict, after `"api_key": cfg.ai.api_key,`: + +```python + "api_keys": cfg.ai.api_keys, +``` + +3d. Add the reconciliation function (place it just above `load_config`): + +```python +def reconcile_keys(ai: Dict[str, Any]) -> None: + """Normalize the API key pool in an ``ai`` config dict, in place. + + Dedupes, drops empties, and keeps the active key first so the invariant + ``api_key == api_keys[0]`` always holds. Seeds the pool from a legacy + single ``api_key`` and lets an env-provided ``api_key`` take priority by + placing it first. + """ + pool = [k for k in dict.fromkeys([ai.get("api_key", "")] + list(ai.get("api_keys") or [])) if k] + ai["api_keys"] = pool + ai["api_key"] = pool[0] if pool else "" +``` + +3e. Call it in `load_config`, immediately after the `allowed_tools` migration block and before `cfg = HackBotConfig(`: + +```python + reconcile_keys(merged.setdefault("ai", {})) +``` + +- [ ] **Step 4: Run tests to verify they pass** + +Run: `pytest tests/test_config.py -v` +Expected: PASS (all existing config tests plus the 5 new ones). + +- [ ] **Step 5: Commit** + +```bash +git add hackbot/config.py tests/test_config.py +git commit -m "$(printf 'feat(config): add DeepSeek api_keys pool with reconciliation\n\nCo-Authored-By: Claude Opus 4.8 ')" +``` + +--- + +## Task 2: Config — pool helpers & key masking + +**Files:** +- Modify: `hackbot/config.py` +- Test: `tests/test_config.py` + +- [ ] **Step 1: Write the failing tests** + +Add to `tests/test_config.py` (extend the `hackbot.config` import to include `add_key`, `remove_key`, `set_active_key`, `mask_key`): + +```python +from hackbot.config import add_key, remove_key, set_active_key, mask_key + + +def test_add_key_appends_and_dedupes(): + cfg = AIConfig(provider="deepseek", api_key="sk-a", api_keys=["sk-a"]) + add_key(cfg, "sk-b") + assert cfg.api_keys == ["sk-a", "sk-b"] + add_key(cfg, "sk-b") # duplicate is a no-op + assert cfg.api_keys == ["sk-a", "sk-b"] + assert cfg.api_key == "sk-a" # active unchanged + + +def test_add_key_seeds_active_when_empty(): + cfg = AIConfig(provider="deepseek", api_key="", api_keys=[]) + add_key(cfg, "sk-a") + assert cfg.api_keys == ["sk-a"] + assert cfg.api_key == "sk-a" + + +def test_remove_active_key_promotes_next(): + cfg = AIConfig(provider="deepseek", api_key="sk-a", api_keys=["sk-a", "sk-b"]) + removed = remove_key(cfg, 0) + assert removed == "sk-a" + assert cfg.api_keys == ["sk-b"] + assert cfg.api_key == "sk-b" + + +def test_remove_nonactive_key_keeps_active(): + cfg = AIConfig(provider="deepseek", api_key="sk-a", api_keys=["sk-a", "sk-b"]) + removed = remove_key(cfg, 1) + assert removed == "sk-b" + assert cfg.api_keys == ["sk-a"] + assert cfg.api_key == "sk-a" + + +def test_remove_key_bad_index_raises(): + cfg = AIConfig(api_keys=["sk-a"]) + with pytest.raises(IndexError): + remove_key(cfg, 5) + + +def test_set_active_key_moves_to_front(): + cfg = AIConfig(provider="deepseek", api_key="sk-a", api_keys=["sk-a", "sk-b"]) + set_active_key(cfg, "sk-b") + assert cfg.api_keys == ["sk-b", "sk-a"] + assert cfg.api_key == "sk-b" + + +def test_set_active_key_new_key_inserts_front(): + cfg = AIConfig(provider="deepseek", api_key="sk-a", api_keys=["sk-a"]) + set_active_key(cfg, "sk-new") + assert cfg.api_keys == ["sk-new", "sk-a"] + assert cfg.api_key == "sk-new" + + +def test_mask_key(): + assert mask_key("sk-1234567890abcd") == "sk-12…abcd" + assert mask_key("short") == "sho…" + assert mask_key("") == "(empty)" +``` + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `pytest tests/test_config.py -k "add_key or remove or set_active or mask" -v` +Expected: FAIL — `ImportError: cannot import name 'add_key'`. + +- [ ] **Step 3: Implement** + +Add to `hackbot/config.py` (just below `reconcile_keys`): + +```python +def add_key(cfg: "AIConfig", key: str) -> None: + """Append *key* to the pool if not already present; keep ``api_key`` valid.""" + key = key.strip() + if not key or key in cfg.api_keys: + return + cfg.api_keys.append(key) + if not cfg.api_key: + cfg.api_key = cfg.api_keys[0] + + +def remove_key(cfg: "AIConfig", index: int) -> str: + """Remove and return the pool entry at *index*. Raises IndexError if invalid. + + If the removed key was the active one, the new active key becomes the new + ``api_keys[0]`` (or empty when the pool is exhausted). + """ + removed = cfg.api_keys.pop(index) # raises IndexError on bad index + cfg.api_key = cfg.api_keys[0] if cfg.api_keys else "" + return removed + + +def set_active_key(cfg: "AIConfig", key: str) -> None: + """Make *key* the active key by moving/inserting it at the front of the pool.""" + key = key.strip() + if not key: + return + if key in cfg.api_keys: + cfg.api_keys.remove(key) + cfg.api_keys.insert(0, key) + cfg.api_key = key + + +def mask_key(key: str) -> str: + """Return a display-safe masked form of an API key.""" + if not key: + return "(empty)" + if len(key) <= 8: + return key[:3] + "…" + return f"{key[:5]}…{key[-4:]}" +``` + +- [ ] **Step 4: Run tests to verify they pass** + +Run: `pytest tests/test_config.py -v` +Expected: PASS. + +- [ ] **Step 5: Commit** + +```bash +git add hackbot/config.py tests/test_config.py +git commit -m "$(printf 'feat(config): add pool helpers and mask_key\n\nCo-Authored-By: Claude Opus 4.8 ')" +``` + +--- + +## Task 3: Engine — failure classifier & `KeyPoolExhaustedError` + +**Files:** +- Modify: `hackbot/core/engine.py` +- Test: `tests/test_engine.py` + +- [ ] **Step 1: Write the failing tests** + +Add to `tests/test_engine.py` (extend the `hackbot.core.engine` import to include `KeyPoolExhaustedError`): + +```python +from hackbot.core.engine import KeyPoolExhaustedError + + +class TestKeyFailureReason: + """AIEngine._key_failure_reason classifies which failures rotate a key.""" + + @pytest.mark.parametrize("message,expected", [ + ("Error code: 401 - Unauthorized", "dead"), + ("Invalid API key provided", "dead"), + ("Error code: 403 Forbidden", "disabled"), + ("Error code: 402 - Insufficient Balance", "out of credit"), + ("Error code: 429 - You exceeded your current quota, insufficient_quota", "out of credit"), + ("Error code: 429 - Rate limit exceeded", None), + ("Error code: 404 - Model not found", None), + ("Error code: 400 - Bad request", None), + ("Connection refused", None), + ("Request timed out", None), + ]) + def test_classifier(self, message, expected): + assert AIEngine._key_failure_reason(Exception(message)) == expected + + def test_classifier_uses_status_code(self): + exc = Exception("payment required") + exc.status_code = 402 + assert AIEngine._key_failure_reason(exc) == "out of credit" +``` + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `pytest tests/test_engine.py -k "KeyFailureReason" -v` +Expected: FAIL — `ImportError: cannot import name 'KeyPoolExhaustedError'`. + +- [ ] **Step 3: Implement** + +In `hackbot/core/engine.py`: + +3a. Change the config import line `from hackbot.config import AIConfig` to: + +```python +from hackbot.config import AIConfig, mask_key +``` + +3b. Add the exception class immediately after the imports (above `# ── Provider Registry`): + +```python +class KeyPoolExhaustedError(RuntimeError): + """Raised when every key in the DeepSeek pool is dead or out of credit.""" +``` + +3c. Add the classifier as a `@staticmethod` on `AIEngine` (place it just above `validate_api_key`): + +```python + @staticmethod + def _key_failure_reason(exc: Exception) -> Optional[str]: + """Classify an API error as a key-rotation reason, or None. + + Returns 'out of credit' | 'dead' | 'disabled' when the *current key* is + unusable, or None for transient / non-key errors (plain rate-limits, + 404 model-not-found, 400, timeouts, connection failures) that rotating + through the pool would not fix. + """ + msg = str(exc).lower() + code = getattr(exc, "status_code", None) + if (code == 402 or "insufficient balance" in msg or "insufficient_quota" in msg + or "exceeded your current quota" in msg): + return "out of credit" + if (code == 401 or "401" in msg or "unauthorized" in msg + or "invalid api key" in msg or "invalid_api_key" in msg): + return "dead" + if code == 403 or "403" in msg or "forbidden" in msg: + return "disabled" + return None +``` + +- [ ] **Step 4: Run tests to verify they pass** + +Run: `pytest tests/test_engine.py -k "KeyFailureReason" -v` +Expected: PASS (12 parametrized cases + status-code case). + +- [ ] **Step 5: Commit** + +```bash +git add hackbot/core/engine.py tests/test_engine.py +git commit -m "$(printf 'feat(engine): add key-failure classifier and KeyPoolExhaustedError\n\nCo-Authored-By: Claude Opus 4.8 ')" +``` + +--- + +## Task 4: Engine — blocking-call failover + +**Files:** +- Modify: `hackbot/core/engine.py` +- Test: `tests/test_engine.py` + +- [ ] **Step 1: Write the failing tests** + +Add to `tests/test_engine.py`: + +```python +class TestBlockingFailover: + @patch("hackbot.core.engine.OpenAI") + def test_rotates_on_out_of_credit(self, mock_openai_cls): + mock_client = MagicMock() + mock_openai_cls.return_value = mock_client + ok = MagicMock() + ok.choices = [MagicMock()] + ok.choices[0].message.content = "hi from key2" + mock_client.chat.completions.create.side_effect = [ + Exception("Error code: 402 - Insufficient Balance"), + ok, + ] + cfg = AIConfig(provider="deepseek", model="deepseek-chat", + api_key="sk-1", api_keys=["sk-1", "sk-2"]) + engine = AIEngine(cfg) + notices = [] + engine.on_notice = notices.append + + result = engine.chat(create_conversation("chat"), stream=False) + + assert result == "hi from key2" + assert cfg.api_key == "sk-2" + assert mock_client.chat.completions.create.call_count == 2 + assert notices and "switched to key #2" in notices[0] + + @patch("hackbot.core.engine.OpenAI") + def test_raises_when_all_keys_exhausted(self, mock_openai_cls): + mock_client = MagicMock() + mock_openai_cls.return_value = mock_client + mock_client.chat.completions.create.side_effect = [ + Exception("Error code: 402 - Insufficient Balance"), + Exception("Error code: 401 - Unauthorized"), + ] + cfg = AIConfig(provider="deepseek", model="deepseek-chat", + api_key="sk-1", api_keys=["sk-1", "sk-2"]) + engine = AIEngine(cfg) + + with pytest.raises(KeyPoolExhaustedError) as info: + engine.chat(create_conversation("chat"), stream=False) + assert "exhausted" in str(info.value).lower() + + @patch("hackbot.core.engine.OpenAI") + def test_no_failover_for_non_deepseek(self, mock_openai_cls): + mock_client = MagicMock() + mock_openai_cls.return_value = mock_client + mock_client.chat.completions.create.side_effect = Exception( + "Error code: 402 - Insufficient Balance" + ) + cfg = AIConfig(provider="openai", model="gpt-4o", + api_key="sk-1", api_keys=["sk-1", "sk-2"]) + engine = AIEngine(cfg) + + with pytest.raises(Exception) as info: + engine.chat(create_conversation("chat"), stream=False) + assert "402" in str(info.value) + assert mock_client.chat.completions.create.call_count == 1 +``` + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `pytest tests/test_engine.py -k "BlockingFailover" -v` +Expected: FAIL — the second `create` is never retried (`call_count == 1`), so `test_rotates_on_out_of_credit` errors out of `_blocking_chat`. + +- [ ] **Step 3: Implement** + +3a. In `AIEngine.__init__`, after `self._setup_client()`, add rotation state: + +```python + self.on_notice: Optional[Callable[[str], None]] = None + self._bad: Dict[str, str] = {} # session-only: key -> failure reason +``` + +3b. Add these members to `AIEngine` (place above `chat`): + +```python + @property + def _failover_enabled(self) -> bool: + return self.config.provider == "deepseek" and len([k for k in self.config.api_keys if k]) > 1 + + @property + def bad_keys(self) -> Dict[str, str]: + """Session-only map of pool keys that have failed → reason.""" + return dict(self._bad) + + def _key_index(self, key: str) -> int: + try: + return self.config.api_keys.index(key) + 1 + except ValueError: + return 0 + + def _next_good_key(self) -> Optional[str]: + for k in self.config.api_keys: + if k and k not in self._bad: + return k + return None + + def _activate(self, key: str) -> None: + self.config.api_key = key + self._setup_client() + + def _exhausted_message(self) -> str: + lines = [f" • {mask_key(k)}: {self._bad.get(k, 'unknown')}" + for k in self.config.api_keys if k] + return ("All DeepSeek keys are exhausted or invalid:\n" + + "\n".join(lines) + + "\n\nAdd a working key with: /key add ") + + def _maybe_rotate(self, exc: Exception) -> bool: + """Rotate to the next good key after a key-related failure. + + Returns True if rotation happened (caller should retry), False if the + error is transient / not key-related or failover is disabled. Raises + KeyPoolExhaustedError when no usable key remains. + """ + if not self._failover_enabled: + return False + reason = self._key_failure_reason(exc) + if reason is None: + return False + bad_key = self.config.api_key + self._bad[bad_key] = reason + nxt = self._next_good_key() + if nxt is None: + raise KeyPoolExhaustedError(self._exhausted_message()) from exc + prev_n = self._key_index(bad_key) + self._activate(nxt) + if self.on_notice: + self.on_notice(f"DeepSeek key #{prev_n} {reason} — switched to key #{self._key_index(nxt)}") + return True +``` + +3c. Replace the body of `_blocking_chat` with a failover loop: + +```python + def _blocking_chat(self, messages: List[Dict[str, str]]) -> str: + while True: + try: + response = self.client.chat.completions.create( + model=self.config.model, + messages=messages, + temperature=self.config.temperature, + max_tokens=self.config.max_tokens, + ) + return response.choices[0].message.content or "" + except Exception as exc: + if not self._maybe_rotate(exc): + raise +``` + +- [ ] **Step 4: Run tests to verify they pass** + +Run: `pytest tests/test_engine.py -k "BlockingFailover" -v` +Expected: PASS (3 tests). + +- [ ] **Step 5: Commit** + +```bash +git add hackbot/core/engine.py tests/test_engine.py +git commit -m "$(printf 'feat(engine): DeepSeek key failover for blocking calls\n\nCo-Authored-By: Claude Opus 4.8 ')" +``` + +--- + +## Task 5: Engine — streaming failover + +**Files:** +- Modify: `hackbot/core/engine.py` +- Test: `tests/test_engine.py` + +- [ ] **Step 1: Write the failing tests** + +Add to `tests/test_engine.py`: + +```python +def _chunk(text): + c = MagicMock() + c.choices = [MagicMock()] + c.choices[0].delta.content = text + return c + + +class TestStreamingFailover: + @patch("hackbot.core.engine.OpenAI") + def test_retries_before_first_token(self, mock_openai_cls): + mock_client = MagicMock() + mock_openai_cls.return_value = mock_client + mock_client.chat.completions.create.side_effect = [ + Exception("Error code: 402 - Insufficient Balance"), + iter([_chunk("hi"), _chunk(" there")]), + ] + cfg = AIConfig(provider="deepseek", model="deepseek-chat", + api_key="sk-1", api_keys=["sk-1", "sk-2"]) + engine = AIEngine(cfg) + tokens = [] + + result = engine.chat(create_conversation("chat"), stream=True, on_token=tokens.append) + + assert result == "hi there" + assert tokens == ["hi", " there"] + assert cfg.api_key == "sk-2" + + @patch("hackbot.core.engine.OpenAI") + def test_no_retry_after_first_token(self, mock_openai_cls): + mock_client = MagicMock() + mock_openai_cls.return_value = mock_client + + def gen(): + yield _chunk("partial") + raise Exception("Error code: 402 - Insufficient Balance") + + mock_client.chat.completions.create.return_value = gen() + cfg = AIConfig(provider="deepseek", model="deepseek-chat", + api_key="sk-1", api_keys=["sk-1", "sk-2"]) + engine = AIEngine(cfg) + tokens = [] + + with pytest.raises(Exception) as info: + engine.chat(create_conversation("chat"), stream=True, on_token=tokens.append) + + assert "402" in str(info.value) + assert tokens == ["partial"] + assert mock_client.chat.completions.create.call_count == 1 +``` + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `pytest tests/test_engine.py -k "StreamingFailover" -v` +Expected: FAIL — `test_retries_before_first_token` raises instead of rotating. + +- [ ] **Step 3: Implement** + +Replace the body of `_stream_chat` with a failover loop that only retries before the first token is emitted: + +```python + def _stream_chat( + self, + messages: List[Dict[str, str]], + on_token: Callable[[str], None], + ) -> str: + while True: + emitted = False + parts: List[str] = [] + try: + stream = self.client.chat.completions.create( + model=self.config.model, + messages=messages, + temperature=self.config.temperature, + max_tokens=self.config.max_tokens, + stream=True, + ) + for chunk in stream: + if chunk.choices and chunk.choices[0].delta.content: + token = chunk.choices[0].delta.content + emitted = True + parts.append(token) + on_token(token) + return "".join(parts) + except Exception as exc: + # Already streamed output — retrying would duplicate it. + if emitted: + raise + if not self._maybe_rotate(exc): + raise + # rotated to a fresh key — retry the request from the top +``` + +- [ ] **Step 4: Run tests to verify they pass** + +Run: `pytest tests/test_engine.py -v` +Expected: PASS (all engine tests, including the 13 existing `validate_api_key` tests). + +- [ ] **Step 5: Commit** + +```bash +git add hackbot/core/engine.py tests/test_engine.py +git commit -m "$(printf 'feat(engine): DeepSeek key failover for streaming calls\n\nCo-Authored-By: Claude Opus 4.8 ')" +``` + +--- + +## Task 6: CLI — `_rebuild_engine` helper & notice wiring + +**Files:** +- Modify: `hackbot/cli.py` + +This refactor introduces one helper that rebuilds the engine *and* wires the rotation notice to `print_warning`, then routes the two existing rebuild sites through it. (`_set_key` is rewritten in Task 7 and will use the same helper.) + +- [ ] **Step 1: Add the helper** + +In `hackbot/cli.py`, add this method to `HackBotApp` (place it just above `_show_config`): + +```python + def _rebuild_engine(self) -> None: + """Rebuild the AI engine from current config and re-wire it to all modes.""" + self.engine = AIEngine(self.config.ai) + self.engine.on_notice = lambda msg: print_warning(msg) + self.chat.engine = self.engine + self.plan.engine = self.engine + if self.agent: + self.agent.engine = self.engine + self.agent.summarizer.engine = self.engine +``` + +- [ ] **Step 2: Wire the notice on the initial engine** + +In `HackBotApp.__init__`, immediately after `self._start_time = time.time()`, add: + +```python + self.engine.on_notice = lambda msg: print_warning(msg) +``` + +- [ ] **Step 3: Route `_set_model` through the helper** + +In `_set_model`, replace these lines: + +```python + self.config.ai.model = model + self.engine = AIEngine(self.config.ai) + # Update all modes with new engine + self.chat.engine = self.engine + self.plan.engine = self.engine + if self.agent: + self.agent.engine = self.engine + self.agent.summarizer.engine = self.engine + save_config(self.config) +``` + +with: + +```python + self.config.ai.model = model + self._rebuild_engine() + save_config(self.config) +``` + +- [ ] **Step 4: Route `_set_provider` through the helper** + +In `_set_provider`, replace these lines: + +```python + self.engine = AIEngine(self.config.ai) + self.chat.engine = self.engine + self.plan.engine = self.engine + if self.agent: + self.agent.engine = self.engine + self.agent.summarizer.engine = self.engine + save_config(self.config) +``` + +with: + +```python + self._rebuild_engine() + save_config(self.config) +``` + +- [ ] **Step 5: Verify nothing regressed** + +Run: `python -c "from hackbot.cli import HackBotApp; from hackbot.config import HackBotConfig; a = HackBotApp(HackBotConfig()); print('on_notice wired:', callable(a.engine.on_notice))"` +Expected: `on_notice wired: True` + +Run: `pytest tests/ -q` +Expected: PASS (no regressions). + +- [ ] **Step 6: Commit** + +```bash +git add hackbot/cli.py +git commit -m "$(printf 'refactor(cli): add _rebuild_engine helper and wire rotation notices\n\nCo-Authored-By: Claude Opus 4.8 ')" +``` + +--- + +## Task 7: CLI — `/key` subcommands, `/keys`, and help text + +**Files:** +- Modify: `hackbot/cli.py`, `hackbot/ui/__init__.py` +- Test: `tests/test_keypool_cli.py` (create) + +- [ ] **Step 1: Write the failing tests** + +Create `tests/test_keypool_cli.py`: + +```python +"""Tests for the DeepSeek key-pool REPL commands.""" + +from unittest.mock import MagicMock, patch + +import pytest + +from hackbot.cli import HackBotApp +from hackbot.config import HackBotConfig + + +def _app(provider="deepseek", keys=None, active=""): + cfg = HackBotConfig() + cfg.ai.provider = provider + cfg.ai.model = "deepseek-chat" if provider == "deepseek" else "gpt-4o" + cfg.ai.api_keys = list(keys or []) + cfg.ai.api_key = active or (cfg.ai.api_keys[0] if cfg.ai.api_keys else "") + return HackBotApp(cfg) + + +@patch("hackbot.cli.save_config") +@patch("hackbot.core.engine.OpenAI") +def test_key_add_appends(mock_openai, mock_save): + mock_client = MagicMock() + mock_openai.return_value = mock_client + mock_client.chat.completions.create.return_value = MagicMock(model="deepseek-chat") + app = _app(keys=["sk-1"], active="sk-1") + + app._set_key("add sk-2") + + assert app.config.ai.api_keys == ["sk-1", "sk-2"] + assert mock_save.called + + +@patch("hackbot.cli.save_config") +@patch("hackbot.core.engine.OpenAI") +def test_key_add_guarded_on_non_deepseek(mock_openai, mock_save): + mock_openai.return_value = MagicMock() + app = _app(provider="openai", keys=["sk-1"], active="sk-1") + + app._set_key("add sk-2") + + assert app.config.ai.api_keys == ["sk-1"] # unchanged + assert not mock_save.called + + +@patch("hackbot.cli.save_config") +@patch("hackbot.core.engine.OpenAI") +def test_key_remove(mock_openai, mock_save): + mock_openai.return_value = MagicMock() + app = _app(keys=["sk-1", "sk-2"], active="sk-1") + + app._set_key("remove 1") + + assert app.config.ai.api_keys == ["sk-2"] + assert app.config.ai.api_key == "sk-2" + + +@patch("hackbot.cli.save_config") +@patch("hackbot.core.engine.OpenAI") +def test_bare_key_sets_active_front(mock_openai, mock_save): + mock_client = MagicMock() + mock_openai.return_value = mock_client + mock_client.chat.completions.create.return_value = MagicMock(model="deepseek-chat") + app = _app(keys=["sk-1", "sk-2"], active="sk-1") + + app._set_key("sk-2") + + assert app.config.ai.api_keys == ["sk-2", "sk-1"] + assert app.config.ai.api_key == "sk-2" + + +@patch("hackbot.cli.save_config") +@patch("hackbot.core.engine.OpenAI") +def test_list_keys_runs(mock_openai, mock_save): + mock_openai.return_value = MagicMock() + app = _app(keys=["sk-1", "sk-2"], active="sk-1") + + assert app._list_keys() is True # does not raise +``` + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `pytest tests/test_keypool_cli.py -v` +Expected: FAIL — `_set_key("add sk-2")` currently treats `"add sk-2"` as a literal key; `_list_keys` does not exist. + +- [ ] **Step 3: Extend the config import in `cli.py`** + +Change the existing `from hackbot.config import (...)` block to add the four new names: + +```python +from hackbot.config import ( + CONFIG_DIR, + HackBotConfig, + AIConfig, + add_key, + detect_platform, + detect_tools, + load_config, + mask_key, + remove_key, + save_config, + set_active_key, +) +``` + +- [ ] **Step 4: Rewrite `_set_key` with subcommand routing** + +Replace the entire existing `_set_key` method with: + +```python + def _set_key(self, args: str) -> bool: + parts = args.strip().split(maxsplit=1) + sub = parts[0].lower() if parts else "" + rest = parts[1].strip() if len(parts) > 1 else "" + + if sub == "add": + return self._key_add(rest) + if sub == "remove": + return self._key_remove(rest) + if sub == "list": + return self._list_keys() + + key = args.strip() + if not key: + print_error( + "Usage: /key Set the active API key\n" + " /key add Add a DeepSeek key to the pool\n" + " /key remove Remove DeepSeek key #n\n" + " /keys List the DeepSeek key pool" + ) + return True + + if self.config.ai.provider == "deepseek": + set_active_key(self.config.ai, key) + else: + self.config.ai.api_key = key + self._rebuild_engine() + + print_info("Validating API key...") + result = self.engine.validate_api_key() + if result["valid"]: + save_config(self.config) + print_success(result["message"]) + else: + print_error(result["message"]) + print_warning("API key saved but may not work. Use /key to set a valid key.") + save_config(self.config) + return True + + def _key_add(self, key: str) -> bool: + if self.config.ai.provider != "deepseek": + print_error("Multi-key pooling is DeepSeek-only — switch with /provider deepseek.") + return True + key = key.strip() + if not key: + print_error("Usage: /key add ") + return True + if key in self.config.ai.api_keys: + print_info(f"Key {mask_key(key)} is already in the pool.") + return True + + print_info("Validating API key...") + probe = AIEngine(AIConfig(provider="deepseek", model=self.config.ai.model, api_key=key)) + result = probe.validate_api_key() + if result["valid"]: + print_success(result["message"]) + else: + print_warning(f"{result['message']} — adding anyway.") + + add_key(self.config.ai, key) + self._rebuild_engine() + save_config(self.config) + print_success(f"Added DeepSeek key {mask_key(key)} (pool size: {len(self.config.ai.api_keys)}).") + return True + + def _key_remove(self, arg: str) -> bool: + if self.config.ai.provider != "deepseek": + print_error("Multi-key pooling is DeepSeek-only — switch with /provider deepseek.") + return True + arg = arg.strip() + if not arg.isdigit(): + print_error("Usage: /key remove (see numbers with /keys)") + return True + try: + removed = remove_key(self.config.ai, int(arg) - 1) + except IndexError: + print_error(f"No key #{arg} in the pool (see /keys).") + return True + self._rebuild_engine() + save_config(self.config) + print_success(f"Removed DeepSeek key {mask_key(removed)}.") + if not self.config.ai.api_keys: + print_warning("Key pool is now empty. Add one with /key add .") + return True + + def _list_keys(self) -> bool: + if self.config.ai.provider != "deepseek": + print_info( + "Multi-key pooling is DeepSeek-only. " + f"Active key: {mask_key(self.config.ai.api_key)}" + ) + return True + keys = self.config.ai.api_keys + if not keys: + print_info("No DeepSeek keys configured. Add one with /key add .") + return True + bad = self.engine.bad_keys + console.print("\n[bold]DeepSeek key pool:[/]") + for i, k in enumerate(keys, 1): + active = "[green]●[/]" if k == self.config.ai.api_key else " " + status = f" [red]✗ {bad[k]}[/]" if k in bad else "" + console.print(f" {active} #{i} {mask_key(k)}{status}") + print_info(f"{len(keys)} key(s). ● = active.") + return True +``` + +- [ ] **Step 5: Register `/keys` in the dispatch table** + +In `_handle_command`, in the `commands = { ... }` dict, add this entry directly after the `"/key": ...` line: + +```python + "/keys": lambda: self._list_keys(), +``` + +- [ ] **Step 6: Add the commands to `/help`** + +In `hackbot/ui/__init__.py`, inside `show_help`, replace this line: + +```python + /key Set API key +``` + +with: + +```python + /key Set API key + /keys List DeepSeek key pool (active / bad) + /key add Add a DeepSeek key to the pool + /key remove Remove DeepSeek key #n from the pool +``` + +- [ ] **Step 7: Run tests to verify they pass** + +Run: `pytest tests/test_keypool_cli.py -v` +Expected: PASS (5 tests). + +- [ ] **Step 8: Commit** + +```bash +git add hackbot/cli.py hackbot/ui/__init__.py tests/test_keypool_cli.py +git commit -m "$(printf 'feat(cli): DeepSeek key pool commands (/key add, /keys, /key remove)\n\nCo-Authored-By: Claude Opus 4.8 ')" +``` + +--- + +## Task 8: Full verification + +**Files:** none (verification only) + +- [ ] **Step 1: Full test suite** + +Run: `pytest tests/ -q` +Expected: PASS — all tests green, including the new config/engine/cli pool tests. + +- [ ] **Step 2: Lint & format** + +Run: `ruff check hackbot/ tests/` +Expected: no errors. If import-order (I) issues appear, run `ruff check --fix hackbot/ tests/` and re-run. + +Run: `black hackbot/ tests/` +Expected: files already formatted (or reformatted cleanly). + +- [ ] **Step 3: Type check (non-blocking, as in CI)** + +Run: `mypy hackbot/ --ignore-missing-imports` +Expected: no new errors introduced by these changes. + +- [ ] **Step 4: Manual REPL smoke test** + +With two real DeepSeek keys (one valid, one anything), run `hackbot` and enter: + +```text +/provider deepseek +/key +/key add +/keys +/key remove 2 +/keys +``` + +Expected: +- `/keys` lists the pool masked, with `●` on the active key. +- `/key remove 2` removes the second key; the follow-up `/keys` shows one key. +- On another provider (e.g. `/provider openai` then `/key add x`), the command prints the DeepSeek-only guard message and does not modify anything. + +- [ ] **Step 5: Final commit (only if lint/format changed files)** + +```bash +git add -A +git commit -m "$(printf 'chore: lint/format pass for DeepSeek key failover\n\nCo-Authored-By: Claude Opus 4.8 ')" +``` + +--- + +## Notes / deviations from spec + +- **`validate_api_key()` is left intact** rather than rewritten to call the new + classifier. Its categorization serves a different purpose (user-facing + validation messages: it reports rate-limits as *valid*, and distinguishes + 404/timeout). `_key_failure_reason` is a focused sibling that reads the same + signals (status code + message markers) for rotation decisions, so the two do + not drift. The 13 existing validation tests stay green. +- **No standalone `_with_failover(do_call)` wrapper.** Blocking and streaming + retries have different shapes (streaming must not retry once a token is + emitted), so the shared rotation logic lives in `_maybe_rotate()`, which both + `_blocking_chat` and `_stream_chat` call. This preserves the spec's + single-source-of-truth intent. +- **CLI tests added** (`tests/test_keypool_cli.py`) beyond the spec's + engine+config testing scope — cheap, no-network coverage of the new commands. +- **No `refresh_pool()` method** (the spec named one). Every CLI pool edit calls + `_rebuild_engine()`, which constructs a fresh `AIEngine` with empty bad-key + state, so a separate refresh method would be dead code. The spec's intent — + the engine reflects pool edits made by a command — is preserved. +``` diff --git a/docs/superpowers/plans/2026-06-22-autonomous-tool-install.md b/docs/superpowers/plans/2026-06-22-autonomous-tool-install.md new file mode 100644 index 0000000..4eaaeae --- /dev/null +++ b/docs/superpowers/plans/2026-06-22-autonomous-tool-install.md @@ -0,0 +1,904 @@ +# Autonomous Security-Tool Installation Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Let HackBot's agent autonomously install missing security tools mid-assessment via a new `install` action, plus a manual `/install` / `hackbot install` surface. + +**Architecture:** A new `ToolInstaller` (`hackbot/core/installer.py`) turns "I need tool X" into a verified install by selecting an available package manager from a curated `TOOL_INSTALL_MAP` and running the constructed argv through the existing hardened `ToolRunner`. The agent gains an `install` action that is allowlist-bounded, respects `safe_mode` confirmation, and feeds results back into the loop. Installs reuse the runner's shell-free execution, sudo authority, and validation — no new privilege path. + +**Tech Stack:** Python 3.9+, `shutil.which` for manager detection, existing `ToolRunner`/`subprocess`, `pytest` with `monkeypatch`/mocks. + +## Global Constraints + +- Minimum Python is **3.9** — no 3.10+ only syntax (no `match`, no `X | Y` type unions in annotations; use `Optional[...]`/`Dict[...]` from `typing`). +- Execution MUST flow through `ToolRunner.execute` — never a new `subprocess` call. Preserve the shell-free boundary (`shell=False`, `shlex`), `BLOCKED_COMMANDS`, allowlist, and centralized `_apply_sudo`. +- Sudo is applied only by the runner. `InstallPlan.needs_sudo` is metadata; actual sudo happens only when the runner has `sudo_mode` (or the process is root). Do not strip/re-add sudo elsewhere. +- Allowlist-bounded by default: only tools in `agent.allowed_tools` AND present in `TOOL_INSTALL_MAP` are installable. `agent.allow_arbitrary_install` (default `False`) relaxes the allowlist check. +- Respect `safe_mode`: when on, an install requires one `on_confirm` y/n; when off (`--no-safe-mode`), installs run unattended. +- Line length 100; code passes `ruff check` and `black`. Branding rules in CLAUDE.md unchanged. +- Tests MUST NOT install real packages — mock `shutil.which`, `resolve_tool_path`, and the runner. + +--- + +### Task 1: Config — install map, allowlist drivers, and opt-in flag + +**Files:** +- Modify: `hackbot/config.py` (DEFAULT_CONFIG `agent.allowed_tools` ~line 119-137; add `TOOL_INSTALL_MAP` near `TOOL_ALIASES` ~line 479; `AgentConfig` ~line 176-184; `save_config` agent dict ~line 392-401) +- Test: `tests/test_config.py` + +**Interfaces:** +- Produces: + - `TOOL_INSTALL_MAP: Dict[str, Dict[str, object]]` — keys are lowercase tool names; each value has an `"order": List[str]` of manager keys plus one entry per manager key (`"apt"`, `"dnf"`, `"pacman"`, `"brew"`, `"pipx"`, `"pip"`, `"go"`) mapping to that manager's package/module string. + - `AgentConfig.allow_arbitrary_install: bool = False` + - Manager binaries `apt-get`, `dnf`, `pacman`, `brew`, `pipx`, `pip` added to `agent.allowed_tools` (`go`, `python3` already present). + +- [ ] **Step 1: Write the failing test** + +Add to `tests/test_config.py`: + +```python +from hackbot.config import ( + TOOL_INSTALL_MAP, + DEFAULT_CONFIG, + AgentConfig, +) + + +def test_install_map_entries_are_well_formed(): + assert "nuclei" in TOOL_INSTALL_MAP + for tool, recipe in TOOL_INSTALL_MAP.items(): + assert tool == tool.lower() + assert "order" in recipe and recipe["order"], f"{tool} missing order" + for mgr in recipe["order"]: + assert mgr in recipe, f"{tool} order names {mgr} with no package" + + +def test_install_managers_are_allowlisted(): + allowed = set(DEFAULT_CONFIG["agent"]["allowed_tools"]) + for mgr in ("apt-get", "dnf", "pacman", "brew", "pipx", "pip", "go"): + assert mgr in allowed, f"{mgr} must be allowlisted to run installs" + + +def test_allow_arbitrary_install_defaults_off(): + assert AgentConfig().allow_arbitrary_install is False + assert DEFAULT_CONFIG["agent"]["allow_arbitrary_install"] is False +``` + +- [ ] **Step 2: Run test to verify it fails** + +Run: `pytest tests/test_config.py::test_install_map_entries_are_well_formed tests/test_config.py::test_install_managers_are_allowlisted tests/test_config.py::test_allow_arbitrary_install_defaults_off -v` +Expected: FAIL — `ImportError: cannot import name 'TOOL_INSTALL_MAP'` (and the other asserts). + +- [ ] **Step 3: Write minimal implementation** + +In `hackbot/config.py`: + +1. Add the six manager binaries to the `DEFAULT_CONFIG["agent"]["allowed_tools"]` list (after `"go"`): + +```python + "go", + # Package managers — used by ToolInstaller as install drivers only. + "apt-get", + "dnf", + "pacman", + "brew", + "pipx", + "pip", + ], +``` + +2. Add `allow_arbitrary_install` to `DEFAULT_CONFIG["agent"]` (after `"allowed_tools"` is fine; put it before the list for readability): + +```python + "agent": { + "auto_confirm": False, + "max_steps": 50, + "timeout": 300, + "safe_mode": True, + "sudo_mode": False, + "sudo_password": "", + "nvd_api_key": "", + "allow_arbitrary_install": False, + "allowed_tools": [ +``` + +3. Add the field to `AgentConfig` (after `nvd_api_key`): + +```python + nvd_api_key: str = "" + allow_arbitrary_install: bool = False + allowed_tools: List[str] = field(default_factory=lambda: DEFAULT_CONFIG["agent"]["allowed_tools"]) +``` + +4. Persist it in `save_config`'s agent dict (after `"nvd_api_key"`): + +```python + "nvd_api_key": cfg.agent.nvd_api_key, + "allow_arbitrary_install": cfg.agent.allow_arbitrary_install, + "allowed_tools": cfg.agent.allowed_tools, +``` + +5. Add `TOOL_INSTALL_MAP` immediately after the `TOOL_ALIASES` dict (before `def resolve_tool_path`): + +```python +# Curated tool -> package recipes for autonomous installation. +# `order` lists preferred managers; ToolInstaller picks the first one available +# on the host. Per-manager values are the package/module identifier for that +# manager. Only tools also present in agent.allowed_tools are installable. +TOOL_INSTALL_MAP: Dict[str, Dict[str, object]] = { + "nuclei": { + "order": ["go", "apt"], + "go": "github.com/projectdiscovery/nuclei/v3/cmd/nuclei@latest", + "apt": "nuclei", + }, + "subfinder": { + "order": ["go"], + "go": "github.com/projectdiscovery/subfinder/v2/cmd/subfinder@latest", + }, + "httpx": { + "order": ["go"], + "go": "github.com/projectdiscovery/httpx/cmd/httpx@latest", + }, + "ffuf": { + "order": ["go", "apt"], + "go": "github.com/ffuf/ffuf/v2@latest", + "apt": "ffuf", + }, + "gobuster": { + "order": ["go", "apt"], + "go": "github.com/OJ/gobuster/v3@latest", + "apt": "gobuster", + }, + "katana": { + "order": ["go"], + "go": "github.com/projectdiscovery/katana/cmd/katana@latest", + }, + "sqlmap": {"order": ["apt", "pipx"], "apt": "sqlmap", "pipx": "sqlmap"}, + "nikto": {"order": ["apt"], "apt": "nikto"}, + "nmap": {"order": ["apt", "dnf", "pacman", "brew"], "apt": "nmap", + "dnf": "nmap", "pacman": "nmap", "brew": "nmap"}, + "whatweb": {"order": ["apt"], "apt": "whatweb"}, + "wpscan": {"order": ["apt"], "apt": "wpscan"}, + "hydra": {"order": ["apt"], "apt": "hydra"}, + "dnsrecon": {"order": ["apt", "pipx"], "apt": "dnsrecon", "pipx": "dnsrecon"}, + "amass": {"order": ["apt", "go"], "apt": "amass", + "go": "github.com/owasp-amass/amass/v4/...@master"}, + "feroxbuster": {"order": ["apt"], "apt": "feroxbuster"}, + "masscan": {"order": ["apt", "dnf", "pacman"], "apt": "masscan", + "dnf": "masscan", "pacman": "masscan"}, +} +``` + +- [ ] **Step 4: Run test to verify it passes** + +Run: `pytest tests/test_config.py -v` +Expected: PASS (all config tests, including the three new ones). + +- [ ] **Step 5: Lint and commit** + +```bash +ruff check hackbot/config.py && black hackbot/config.py tests/test_config.py +git add hackbot/config.py tests/test_config.py +git commit -m "feat(config): add TOOL_INSTALL_MAP, allow_arbitrary_install flag, install drivers" +``` + +--- + +### Task 2: ToolInstaller — manager detection and install planning + +**Files:** +- Create: `hackbot/core/installer.py` +- Test: `tests/test_installer.py` + +**Interfaces:** +- Consumes: `TOOL_INSTALL_MAP`, `resolve_tool_path` (from `hackbot.config`); `ToolRunner`, `ToolResult` (from `hackbot.core.runner`). +- Produces: + - `@dataclass InstallPlan(tool: str, manager: str, package: str, command: List[str], needs_sudo: bool)` + - `@dataclass InstallResult(tool: str, success: bool, manager: str = "", path: Optional[str] = None, message: str = "", stdout: str = "", stderr: str = "")` + - `class ToolInstaller(runner: ToolRunner, install_map: Optional[Dict] = None)` with: + - `available_managers() -> Dict[str, str]` (logical manager key → binary path; cached) + - `plan_install(tool: str) -> Optional[InstallPlan]` + - `install(plan: InstallPlan) -> InstallResult` (Task 3) + - Module constant `_SUDO_MANAGERS = {"apt", "dnf", "pacman"}`. + +- [ ] **Step 1: Write the failing test** + +Create `tests/test_installer.py`: + +```python +import pytest + +from hackbot.core import installer as inst +from hackbot.core.installer import ToolInstaller, InstallPlan + + +def _installer(monkeypatch, present_binaries, install_map=None): + """ToolInstaller whose available managers are exactly `present_binaries`.""" + def fake_which(binary): + return f"/usr/bin/{binary}" if binary in present_binaries else None + monkeypatch.setattr(inst.shutil, "which", fake_which) + return ToolInstaller(runner=object(), install_map=install_map) + + +def test_available_managers_detects_present_binaries(monkeypatch): + ti = _installer(monkeypatch, {"apt-get", "go"}) + mgrs = ti.available_managers() + assert mgrs.keys() == {"apt", "go"} + assert mgrs["apt"] == "/usr/bin/apt-get" + + +def test_plan_install_picks_first_available_in_order(monkeypatch): + # nuclei order is ["go", "apt"]; only apt present -> apt chosen + ti = _installer(monkeypatch, {"apt-get"}) + plan = ti.plan_install("nuclei") + assert plan is not None + assert plan.manager == "apt" + assert plan.command == ["apt-get", "install", "-y", "nuclei"] + assert plan.needs_sudo is True + + +def test_plan_install_prefers_go_when_available(monkeypatch): + ti = _installer(monkeypatch, {"go", "apt-get"}) + plan = ti.plan_install("nuclei") + assert plan.manager == "go" + assert plan.command[0] == "go" and plan.command[1] == "install" + assert plan.needs_sudo is False + + +def test_plan_install_unmapped_tool_returns_none(monkeypatch): + ti = _installer(monkeypatch, {"apt-get"}) + assert ti.plan_install("definitely-not-a-tool") is None + + +def test_plan_install_no_available_manager_returns_none(monkeypatch): + # subfinder only ships via go; if go absent -> None + ti = _installer(monkeypatch, {"apt-get"}) + assert ti.plan_install("subfinder") is None + + +def test_plan_install_is_case_insensitive(monkeypatch): + ti = _installer(monkeypatch, {"apt-get"}) + assert ti.plan_install("NIKTO").manager == "apt" + + +def test_pacman_command_uses_noconfirm(monkeypatch): + ti = _installer(monkeypatch, {"pacman"}) + plan = ti.plan_install("nmap") + assert plan.command == ["pacman", "-S", "--noconfirm", "nmap"] +``` + +- [ ] **Step 2: Run test to verify it fails** + +Run: `pytest tests/test_installer.py -v` +Expected: FAIL — `ModuleNotFoundError: No module named 'hackbot.core.installer'`. + +- [ ] **Step 3: Write minimal implementation** + +Create `hackbot/core/installer.py`: + +```python +"""Autonomous installation of missing security tools. + +ToolInstaller turns a logical tool name into a verified install by choosing an +available package manager from config.TOOL_INSTALL_MAP and executing the +constructed argv through the existing ToolRunner (shell-free, sudo-aware). +""" + +from __future__ import annotations + +import shutil +from dataclasses import dataclass +from typing import Dict, List, Optional + +from hackbot.config import TOOL_INSTALL_MAP, resolve_tool_path + +# Logical manager key -> binary probed via shutil.which. +_MANAGER_BINARY: Dict[str, str] = { + "apt": "apt-get", + "dnf": "dnf", + "pacman": "pacman", + "brew": "brew", + "pipx": "pipx", + "pip": "pip", + "go": "go", +} + +# System managers that require root for a system-wide install. +_SUDO_MANAGERS = {"apt", "dnf", "pacman"} + + +@dataclass +class InstallPlan: + tool: str + manager: str + package: str + command: List[str] + needs_sudo: bool + + +@dataclass +class InstallResult: + tool: str + success: bool + manager: str = "" + path: Optional[str] = None + message: str = "" + stdout: str = "" + stderr: str = "" + + +def _build_command(manager: str, package: str) -> List[str]: + if manager == "apt": + return ["apt-get", "install", "-y", package] + if manager == "dnf": + return ["dnf", "install", "-y", package] + if manager == "pacman": + return ["pacman", "-S", "--noconfirm", package] + if manager == "brew": + return ["brew", "install", package] + if manager == "pipx": + return ["pipx", "install", package] + if manager == "pip": + return ["pip", "install", package] + if manager == "go": + return ["go", "install", package] + raise ValueError(f"unknown manager: {manager}") + + +class ToolInstaller: + def __init__(self, runner, install_map: Optional[Dict] = None): + self.runner = runner + self.install_map = install_map if install_map is not None else TOOL_INSTALL_MAP + self._available: Optional[Dict[str, str]] = None + + def available_managers(self) -> Dict[str, str]: + if self._available is None: + found: Dict[str, str] = {} + for mgr, binary in _MANAGER_BINARY.items(): + path = shutil.which(binary) + if path: + found[mgr] = path + self._available = found + return self._available + + def plan_install(self, tool: str) -> Optional[InstallPlan]: + recipe = self.install_map.get(tool.lower()) + if not recipe: + return None + available = self.available_managers() + order = recipe.get("order") or [k for k in recipe if k != "order"] + for mgr in order: + package = recipe.get(mgr) + if package and mgr in available: + return InstallPlan( + tool=tool, + manager=mgr, + package=str(package), + command=_build_command(mgr, str(package)), + needs_sudo=mgr in _SUDO_MANAGERS, + ) + return None +``` + +- [ ] **Step 4: Run test to verify it passes** + +Run: `pytest tests/test_installer.py -v` +Expected: PASS (all 7 tests). + +- [ ] **Step 5: Lint and commit** + +```bash +ruff check hackbot/core/installer.py && black hackbot/core/installer.py tests/test_installer.py +git add hackbot/core/installer.py tests/test_installer.py +git commit -m "feat(installer): ToolInstaller manager detection and install planning" +``` + +--- + +### Task 3: ToolInstaller.install — execute via runner and verify + +**Files:** +- Modify: `hackbot/core/installer.py` +- Test: `tests/test_installer.py` + +**Interfaces:** +- Consumes: `InstallPlan` (Task 2); `ToolResult` (from `hackbot.core.runner`); `resolve_tool_path`. +- Produces: `ToolInstaller.install(plan: InstallPlan) -> InstallResult` — runs `plan.command` (joined to a string) through `self.runner.execute(...)`, then verifies via `resolve_tool_path(plan.tool)`. Success is `runner result success AND binary now resolves`. + +- [ ] **Step 1: Write the failing test** + +Append to `tests/test_installer.py`: + +```python +from hackbot.core.runner import ToolResult +from hackbot.core.installer import InstallPlan + + +class _FakeRunner: + def __init__(self, success=True): + self._success = success + self.commands = [] + + def execute(self, command, tool_name="", explanation=""): + self.commands.append(command) + return ToolResult( + tool=tool_name, command=command, + stdout="installed" if self._success else "", + stderr="" if self._success else "E: permission denied", + return_code=0 if self._success else 1, + duration=0.1, success=self._success, + ) + + +_PLAN = InstallPlan(tool="nuclei", manager="go", + command=["go", "install", "x@latest"], needs_sudo=False) + + +def test_install_success_when_runner_ok_and_binary_resolves(monkeypatch): + monkeypatch.setattr(inst, "resolve_tool_path", lambda t: "/usr/bin/nuclei") + runner = _FakeRunner(success=True) + ti = ToolInstaller(runner=runner) + res = ti.install(_PLAN) + assert res.success is True + assert res.path == "/usr/bin/nuclei" + assert runner.commands == ["go install x@latest"] + + +def test_install_fails_when_runner_returns_nonzero(monkeypatch): + monkeypatch.setattr(inst, "resolve_tool_path", lambda t: None) + ti = ToolInstaller(runner=_FakeRunner(success=False)) + res = ti.install(_PLAN) + assert res.success is False + assert "permission denied" in res.stderr + + +def test_install_fails_when_binary_still_missing(monkeypatch): + # Runner reports success but the binary is not on PATH afterward. + monkeypatch.setattr(inst, "resolve_tool_path", lambda t: None) + ti = ToolInstaller(runner=_FakeRunner(success=True)) + res = ti.install(_PLAN) + assert res.success is False + assert res.path is None +``` + +- [ ] **Step 2: Run test to verify it fails** + +Run: `pytest tests/test_installer.py -k install -v` +Expected: FAIL — `AttributeError: 'ToolInstaller' object has no attribute 'install'`. + +- [ ] **Step 3: Write minimal implementation** + +Add the `install` method to `ToolInstaller` in `hackbot/core/installer.py`: + +```python + def install(self, plan: InstallPlan) -> InstallResult: + command_str = " ".join(plan.command) + result = self.runner.execute( + command_str, + tool_name=plan.manager, + explanation=f"Install {plan.tool}", + ) + path = resolve_tool_path(plan.tool) + success = result.success and path is not None + if success: + message = f"Installed {plan.tool} via {plan.manager} ({path})" + elif result.success: + message = ( + f"{plan.manager} reported success but {plan.tool} is still not on PATH" + ) + else: + message = f"Install of {plan.tool} via {plan.manager} failed" + return InstallResult( + tool=plan.tool, + success=success, + manager=plan.manager, + path=path, + message=message, + stdout=(result.stdout or "")[:2000], + stderr=(result.stderr or "")[:2000], + ) +``` + +- [ ] **Step 4: Run test to verify it passes** + +Run: `pytest tests/test_installer.py -v` +Expected: PASS (all installer tests). + +- [ ] **Step 5: Lint and commit** + +```bash +ruff check hackbot/core/installer.py && black hackbot/core/installer.py tests/test_installer.py +git add hackbot/core/installer.py tests/test_installer.py +git commit -m "feat(installer): execute install via runner and verify binary resolves" +``` + +--- + +### Task 4: Agent `install` action — parse and dispatch + +**Files:** +- Modify: `hackbot/modes/agent.py` (`_parse_actions` allowed set ~line 717; `_process_actions_loop` dispatch ~line 591-645; `__init__` ~line 154-176 to construct the installer) +- Test: `tests/test_modes.py` + +**Interfaces:** +- Consumes: `ToolInstaller`, `InstallResult` (Task 3); `resolve_tool_path` (from `hackbot.config`); `self.runner` (ToolRunner — has `is_tool_allowed`, `on_confirm`); `self.config.agent.allow_arbitrary_install`, `self.config.agent.safe_mode`. +- Produces: + - `"install"` added to the `allowed_actions` set in `_parse_actions`. + - `AgentMode._process_install_action(self, action: Dict[str, Any]) -> str` — returns a human/AI-readable result string. + - `self.installer: ToolInstaller` created in `__init__`. + - `_process_actions_loop` handles `atype == "install"` by appending `_process_install_action(action)` to `results_text`. + +- [ ] **Step 1: Write the failing test** + +Add to `tests/test_modes.py` (reuse existing AgentMode construction helpers in that file; if none, build via `AgentMode(engine=, config=load_config())`). The tests below drive `_process_install_action` directly with monkeypatched dependencies: + +```python +import hackbot.modes.agent as agent_mod + + +def _agent(monkeypatch, tmp_path): + """Minimal AgentMode with a stub engine; safe for unit-testing helpers.""" + from hackbot.config import load_config + cfg = load_config() + cfg.agent.safe_mode = False # default off for these unit tests + eng = type("E", (), {"chat": lambda self, *a, **k: ""})() + return AgentMode(engine=eng, config=cfg) + + +def test_parse_actions_recognizes_install(monkeypatch, tmp_path): + a = _agent(monkeypatch, tmp_path) + actions = a._parse_actions('{"action": "install", "tool": "nuclei"}') + assert actions == [{"action": "install", "tool": "nuclei"}] + + +def test_install_action_already_installed_short_circuits(monkeypatch, tmp_path): + a = _agent(monkeypatch, tmp_path) + monkeypatch.setattr(agent_mod, "resolve_tool_path", lambda t: "/usr/bin/nuclei") + msg = a._process_install_action({"action": "install", "tool": "nuclei"}) + assert "already installed" in msg.lower() + + +def test_install_action_rejects_non_allowlisted_tool(monkeypatch, tmp_path): + a = _agent(monkeypatch, tmp_path) + a.config.agent.allow_arbitrary_install = False + monkeypatch.setattr(agent_mod, "resolve_tool_path", lambda t: None) + msg = a._process_install_action({"action": "install", "tool": "evilpkg"}) + assert "allowlist" in msg.lower() + + +def test_install_action_unmapped_reports_no_recipe(monkeypatch, tmp_path): + a = _agent(monkeypatch, tmp_path) + monkeypatch.setattr(agent_mod, "resolve_tool_path", lambda t: None) + # gcc is allowlisted but has no install recipe -> plan is None + monkeypatch.setattr(a.runner, "is_tool_allowed", lambda t: True) + msg = a._process_install_action({"action": "install", "tool": "gcc"}) + assert "no install recipe" in msg.lower() + + +def test_install_action_runs_installer_and_reports(monkeypatch, tmp_path): + a = _agent(monkeypatch, tmp_path) + monkeypatch.setattr(agent_mod, "resolve_tool_path", lambda t: None) + monkeypatch.setattr(a.runner, "is_tool_allowed", lambda t: True) + + from hackbot.core.installer import InstallPlan, InstallResult + plan = InstallPlan("nuclei", "go", ["go", "install", "x"], False) + monkeypatch.setattr(a.installer, "plan_install", lambda t: plan) + monkeypatch.setattr( + a.installer, "install", + lambda p: InstallResult("nuclei", True, "go", "/usr/bin/nuclei", + "Installed nuclei via go (/usr/bin/nuclei)"), + ) + msg = a._process_install_action({"action": "install", "tool": "nuclei"}) + assert "installed nuclei" in msg.lower() + + +def test_install_action_safe_mode_decline(monkeypatch, tmp_path): + a = _agent(monkeypatch, tmp_path) + a.config.agent.safe_mode = True + a.runner.on_confirm = lambda cmd, reason: False # user declines + monkeypatch.setattr(agent_mod, "resolve_tool_path", lambda t: None) + monkeypatch.setattr(a.runner, "is_tool_allowed", lambda t: True) + from hackbot.core.installer import InstallPlan + plan = InstallPlan("nuclei", "go", ["go", "install", "x"], False) + monkeypatch.setattr(a.installer, "plan_install", lambda t: plan) + called = {"installed": False} + monkeypatch.setattr(a.installer, "install", + lambda p: called.__setitem__("installed", True)) + msg = a._process_install_action({"action": "install", "tool": "nuclei"}) + assert "declined" in msg.lower() + assert called["installed"] is False +``` + +- [ ] **Step 2: Run test to verify it fails** + +Run: `pytest tests/test_modes.py -k install -v` +Expected: FAIL — `'install'` not parsed / `AgentMode` has no `_process_install_action` / no `installer` attribute. + +- [ ] **Step 3: Write minimal implementation** + +In `hackbot/modes/agent.py`: + +1. Ensure the import line for config includes `resolve_tool_path` (it imports from `hackbot.config` already; add `resolve_tool_path` to that import) and add the installer import near the top with the other `hackbot.core` imports: + +```python +from hackbot.config import resolve_tool_path +from hackbot.core.installer import ToolInstaller, InstallResult +``` +(If `hackbot.config` is already imported with specific names, append `resolve_tool_path` to that existing import instead of adding a duplicate line.) + +2. Construct the installer in `__init__`, right after the `self.runner = ToolRunner(...)` block (~line 163): + +```python + self.installer = ToolInstaller(self.runner) +``` + +3. Add `"install"` to the `allowed_actions` set in `_parse_actions` (~line 717): + +```python + allowed_actions = {"execute", "finding", "script", "complete", "generate_report", "fuzz", "analyze_anomaly", "chain_exploits", "install"} +``` + +4. Add the dispatch branch in `_process_actions_loop` — place it alongside the other `elif atype == ...` branches (e.g. after the `chain_exploits` branch ~line 645): + +```python + elif atype == "install": + install_result = self._process_install_action(action) + results_text.append(install_result) +``` + +5. Add the handler method (place it near `_execute_action`): + +```python + def _process_install_action(self, action: Dict[str, Any]) -> str: + """Autonomously install a missing security tool (allowlist-bounded).""" + tool = (action.get("tool") or "").strip() + if not tool: + return "**[install]** No tool specified." + + # Allowlist bound (unless the user opted into arbitrary installs). + if not self.config.agent.allow_arbitrary_install and not self.runner.is_tool_allowed(tool): + return ( + f"**[install]** `{tool}` is not in the allowlist; cannot auto-install. " + f"Use a different tool or ask the user to install it manually." + ) + + if resolve_tool_path(tool): + return f"**[install]** `{tool}` is already installed — proceed to use it." + + plan = self.installer.plan_install(tool) + if plan is None: + return ( + f"**[install]** No install recipe for `{tool}` on this system " + f"(unmapped tool or no supported package manager available). " + f"Try a different tool." + ) + + # Respect safe_mode: one confirmation before installing. + if self.config.agent.safe_mode and not self.config.agent.auto_confirm: + confirm = getattr(self.runner, "on_confirm", None) + cmd_str = " ".join(plan.command) + reason = f"INSTALL {tool} via {plan.manager}" + if confirm and not confirm(cmd_str, reason): + return f"**[install]** User declined installation of `{tool}`." + + result: InstallResult = self.installer.install(plan) + if result.success: + return f"**[install]** SUCCESS — {result.message}. You may now use `{tool}`." + detail = (result.stderr or result.stdout or "").strip()[:500] + sudo_hint = "" + if plan.needs_sudo and not self.runner.sudo_mode: + sudo_hint = " (system installs need --sudo / sudo_mode; it appears disabled)" + return f"**[install]** FAILED — {result.message}{sudo_hint}.\n```\n{detail}\n```" +``` + +- [ ] **Step 4: Run test to verify it passes** + +Run: `pytest tests/test_modes.py -k install -v` +Expected: PASS (6 install tests). + +- [ ] **Step 5: Run the full agent/runner suite for regressions** + +Run: `pytest tests/test_modes.py tests/test_runner.py -v` +Expected: PASS (no regressions in existing agent behavior). + +- [ ] **Step 6: Lint and commit** + +```bash +ruff check hackbot/modes/agent.py && black hackbot/modes/agent.py tests/test_modes.py +git add hackbot/modes/agent.py tests/test_modes.py +git commit -m "feat(agent): add allowlist-bounded install action with safe_mode gate" +``` + +--- + +### Task 5: Document the `install` action in the agent system prompt + +**Files:** +- Modify: `hackbot/core/engine.py` (agent action documentation block, ~line 254-260, after the `chain_exploits`/`active_scan` examples) +- Test: `tests/test_engine.py` + +**Interfaces:** +- Consumes: nothing new. +- Produces: the agent system prompt now describes the `install` action so the LLM emits it when a tool run returns `Tool not found`. + +- [ ] **Step 1: Write the failing test** + +Add to `tests/test_engine.py` (use the existing prompt-construction entry point in that file — locate how other tests obtain the agent system prompt, e.g. `AIEngine.get_system_prompt("agent", ...)` or the module-level prompt constant, and mirror it): + +```python +def test_agent_prompt_documents_install_action(): + from hackbot.core.engine import AIEngine + prompt = AIEngine.build_agent_prompt() # match the actual accessor used elsewhere in this file + assert '"action": "install"' in prompt + assert "Tool not found" in prompt +``` + +> Implementer note: `tests/test_engine.py` already exercises the agent prompt. Use whatever accessor those existing tests use to fetch the agent system prompt instead of `build_agent_prompt` if the name differs — the assertion content stays the same. + +- [ ] **Step 2: Run test to verify it fails** + +Run: `pytest tests/test_engine.py::test_agent_prompt_documents_install_action -v` +Expected: FAIL — `"action": "install"` not present in the prompt. + +- [ ] **Step 3: Write minimal implementation** + +In `hackbot/core/engine.py`, add a numbered subsection to the agent prompt action documentation (after the `active_scan` block, before `ZERO-DAY` content ends / wherever the action list continues): + +```python +7. **Autonomous Tool Installation** — If a tool you need is not installed (a command + returns "Tool not found"), you can install it autonomously: + ```json + {"action": "install", "tool": "", "explanation": ""} + ``` + Only allowlisted security tools with a known install recipe can be installed. + When safe_mode is on, the user confirms each install. After a successful install, + re-run the original command. If the install fails or no recipe exists, adapt and + use a different tool. +``` + +(Renumber if `7` collides with an existing item; the content is what matters.) + +- [ ] **Step 4: Run test to verify it passes** + +Run: `pytest tests/test_engine.py -v` +Expected: PASS. + +- [ ] **Step 5: Lint and commit** + +```bash +ruff check hackbot/core/engine.py && black hackbot/core/engine.py tests/test_engine.py +git add hackbot/core/engine.py tests/test_engine.py +git commit -m "docs(engine): document install action in agent system prompt" +``` + +--- + +### Task 6: Manual surface — `/install` slash command and `hackbot install` subcommand + +**Files:** +- Modify: `hackbot/cli.py` (command table ~line 181-220 add `/install`; add `_install_tool` method near `_show_tools` ~line 411; add Click `install` subcommand near the `tools` command ~line 2630) +- Test: `tests/test_cli.py` + +**Interfaces:** +- Consumes: `AgentMode` (already imported in cli), its `.installer`, `.runner`, `_process_install_action`; `HackBotApp._on_confirm` (~line 151). +- Produces: + - `HackBotApp._install_tool(self, args: str) -> bool` — REPL handler; ensures an agent exists (same lazy-create pattern as `_run_command` ~line 963-971), then delegates to `self.agent._process_install_action({"action": "install", "tool": args.strip()})` and prints the returned string. Usage error when `args` empty. + - `"/install": lambda: self._install_tool(args)` entry in the command table. + - Click `install` subcommand (`hackbot install `) that builds a `HackBotApp` and calls `_install_tool`. + +- [ ] **Step 1: Write the failing test** + +Add to `tests/test_cli.py` (mirror the construction style already used there for `HackBotApp`): + +```python +def test_install_tool_requires_argument(make_app): + app = make_app() # use the existing HackBotApp fixture/factory in this file + assert app._install_tool("") is True # returns True (stay in REPL), prints usage + + +def test_install_tool_delegates_to_agent(monkeypatch, make_app): + app = make_app() + captured = {} + + class _StubAgent: + def _process_install_action(self, action): + captured["action"] = action + return "**[install]** SUCCESS — Installed nuclei" + + app.agent = _StubAgent() + app.mode = "agent" + result = app._install_tool("nuclei") + assert result is True + assert captured["action"] == {"action": "install", "tool": "nuclei"} +``` + +> Implementer note: `tests/test_cli.py` already constructs `HackBotApp` for other command tests. Use that existing helper/fixture (named here `make_app`); do not invent a new construction path. If the file builds the app inline, build it inline the same way. + +- [ ] **Step 2: Run test to verify it fails** + +Run: `pytest tests/test_cli.py -k install -v` +Expected: FAIL — `HackBotApp` has no attribute `_install_tool`. + +- [ ] **Step 3: Write minimal implementation** + +In `hackbot/cli.py`: + +1. Add to the `commands` dict (after `"/tools"` ~line 193): + +```python + "/install": lambda: self._install_tool(args), +``` + +2. Add the method (after `_show_tools` ~line 414): + +```python + def _install_tool(self, args: str) -> bool: + tool = args.strip() + if not tool: + print_error("Usage: /install ") + return True + if self.mode != "agent" or not self.agent: + self.agent = AgentMode( + engine=self.engine, + config=self.config, + on_step=self._on_agent_step, + on_confirm=self._on_confirm, + on_output=self._on_tool_output, + ) + result = self.agent._process_install_action({"action": "install", "tool": tool}) + console.print(Markdown(result)) + return True +``` + +3. Add the Click subcommand (after the `tools` command ~line 2637): + +```python +@main.command(name="install") +@click.argument("tool") +@click.pass_context +def install_cmd(ctx, tool): + """Install a security tool that isn't present.""" + config = ctx.obj["config"] + app = HackBotApp(config) + app._install_tool(tool) +``` + +- [ ] **Step 4: Run test to verify it passes** + +Run: `pytest tests/test_cli.py -k install -v` +Expected: PASS. + +- [ ] **Step 5: Full suite + smoke test** + +```bash +pytest tests/ -q +hackbot --version +``` +Expected: full suite passes; version prints. + +- [ ] **Step 6: Lint and commit** + +```bash +ruff check hackbot/cli.py && black hackbot/cli.py tests/test_cli.py +git add hackbot/cli.py tests/test_cli.py +git commit -m "feat(cli): add /install command and hackbot install subcommand" +``` + +--- + +## Self-Review + +**Spec coverage:** +- `installer.py` / `ToolInstaller` + `InstallPlan`/`InstallResult` → Tasks 2, 3. ✓ +- `TOOL_INSTALL_MAP` + layered managers (apt/dnf/pacman/brew/pipx/pip/go) → Task 1 (map) + Task 2 (`_build_command`, `_MANAGER_BINARY`). ✓ +- `install` agent action (parse + dispatch, allowlist, already-installed, plan, safe_mode confirm, feed-back) → Task 4. ✓ +- Engine prompt documents the action + "Tool not found" guidance → Task 5. ✓ +- Security: execution via `ToolRunner`, managers allowlisted as drivers, sudo via `_apply_sudo` only (`needs_sudo` is metadata + honest sudo hint) → Tasks 1, 3, 4. ✓ +- `allow_arbitrary_install` flag default off → Task 1; enforced in Task 4. ✓ +- Manual `/install` + `hackbot install` → Task 6. ✓ +- Tests with mocks, no real installs → every task. ✓ +- Error handling (unmapped, nonzero exit, binary still missing, declined, no managers) → Tasks 3 + 4 tests. ✓ + +**Placeholder scan:** No TBD/TODO; every code step shows full code. Two "implementer notes" (Tasks 5, 6) point at existing test accessors/fixtures rather than guessing their exact names — the assertion content and behavior are fully specified, so this is a lookup, not an unfinished spec. + +**Type consistency:** `InstallPlan(tool, manager, package, command, needs_sudo)` and `InstallResult(tool, success, manager, path, message, stdout, stderr)` are used identically across Tasks 2–4 and tests. `plan_install`/`install`/`available_managers`/`_process_install_action` signatures match between definition and callers. `_MANAGER_BINARY` keys (`apt`,`dnf`,…) align with `TOOL_INSTALL_MAP` per-manager keys and `_SUDO_MANAGERS`. diff --git a/docs/superpowers/specs/2026-06-21-deepseek-multi-key-failover-design.md b/docs/superpowers/specs/2026-06-21-deepseek-multi-key-failover-design.md new file mode 100644 index 0000000..3c6d6c1 --- /dev/null +++ b/docs/superpowers/specs/2026-06-21-deepseek-multi-key-failover-design.md @@ -0,0 +1,212 @@ +# DeepSeek Multi-Key Failover — Design + +- **Date:** 2026-06-21 +- **Status:** Approved (pending spec review) +- **Scope:** DeepSeek provider only + +## Summary + +Let users store a **pool of DeepSeek API keys** so that when the active key +goes **bad** (revoked/invalid) or **runs out** (insufficient balance / quota), +HackBot automatically rotates to the next working key and **retries the same +request transparently**, with a brief notice. Keys are managed from the REPL +(`/key add`, `/keys`, `/key remove`). All other providers behave exactly as +they do today. + +## Motivation + +A single DeepSeek key is a single point of failure: when it is revoked or its +balance hits zero, every request fails until the user notices and manually +swaps the key. A small key pool with automatic failover keeps an assessment or +chat session running without interruption. + +## Goals + +- Store multiple DeepSeek keys, ordered by priority. +- On a rotation-worthy failure, switch to the next untried key and retry the + same request with no user action. +- Manage the pool interactively: add, list, remove. +- Full backward compatibility with existing single-key configs. + +## Non-goals (out of scope) + +- Cross-provider failover (DeepSeek → Groq, etc.). +- Multi-key support for any provider other than DeepSeek. +- Rotating on transient errors (plain rate-limits, network/timeout, model-not-found). +- A GUI management surface for the pool (failover still works in the GUI because + it is engine-level; only the *management UI* is deferred — see Future Work). +- Encrypted key storage (keys remain plaintext in `config.yaml`, unchanged from + today's `api_key`). +- Changing the `hackbot setup` command (it stays single-key and seeds the pool). + +## Requirements (decisions captured during brainstorming) + +1. **Failover scope:** more keys for the **same provider** (DeepSeek), not other providers. +2. **Trigger behavior:** **automatic** rotation **with retry** of the failed request. +3. **What counts as failure worth rotating:** key is **dead** (`401`/`403`) or + **out of credit** (`402` Insufficient Balance, or `429` with + `insufficient_quota`). Plain `429` rate-limits, `404`, `400`, and + network/timeout errors do **not** rotate. +4. **Storage:** a single DeepSeek key pool (DeepSeek-only feature; no per-provider map). +5. **Management:** REPL commands `/key add `, `/keys`, `/key remove `. +6. **Bad-key memory:** session-only. Every run starts fresh with the full pool + (quotas refill, balances get topped up). + +## Design + +### 1. Data model & config (`config.py`) + +- Add **`api_keys: list[str]`** to `AIConfig` (default `[]`): the ordered pool, + highest priority first. Update `DEFAULT_CONFIG["ai"]` and `save_config()` to + round-trip it. +- **Invariant:** `api_key` always mirrors the **active** key (`pool[0]` at + startup). Every existing reader of `config.ai.api_key` keeps working + unchanged — this is what preserves backward compatibility. +- **Reconciliation** at the end of `load_config()`, after env-var overrides so + env-provided keys still win: + + ```python + keys = [k for k in dict.fromkeys([api_key] + api_keys) if k] # dedupe, active-first + merged["ai"]["api_keys"] = keys + merged["ai"]["api_key"] = keys[0] if keys else "" + ``` + + Effects: an old single-key config auto-seeds the pool `[that key]`; a pool + config sets the active key to `pool[0]`; an env-provided key is prepended as + the active key. No migration script required. +- **Pure pool helpers** (testable without the CLI), operating on `AIConfig` and + maintaining the dedupe + active-first invariant: + - `add_key(cfg, key) -> None` + - `remove_key(cfg, index) -> str` (returns removed key; raises on bad index) + - `set_active_key(cfg, key) -> None` (moves/inserts key to front, updates `api_key`) + +### 2. Engine failover (`core/engine.py`) + +- **Gating:** failover engages only when `config.provider == "deepseek"`. Any + other provider performs a single attempt and raises on error — today's + behavior, unchanged. +- **Shared error classifier:** extract the categorization currently inside + `validate_api_key()` into `_key_failure_reason(exc) -> str | None`, used by + both validation and failover so they cannot drift: + + | Condition | Result | + |---|---| + | `status_code == 401` | `"dead"` → rotate | + | `status_code == 403` | `"disabled"` → rotate | + | `status_code == 402` (Insufficient Balance) | `"out of credit"` → rotate | + | `429` with `insufficient_quota` / `insufficient balance` in message | `"out of credit"` → rotate | + | plain `429` rate-limit, `404`, `400`, timeout/connection | `None` → do not rotate, re-raise | + + Classification reads `getattr(exc, "status_code", None)` and falls back to + scanning `str(exc)` for the quota markers. +- **Engine runtime state:** `self._bad: dict[str, str]` (key → reason) tracks + session-bad keys; the active key is `config.api_key`. A `refresh_pool()` method + re-reads the config after the pool is mutated by a command. +- **`_with_failover(do_call, *, on_notice)`** wraps the completion call: + 1. Try `do_call(self.client)`. + 2. On exception, compute `reason = _key_failure_reason(exc)`. If `reason is + None` or failover is not enabled → re-raise. + 3. Mark the active key bad (`self._bad`), pick the next key not in `self._bad`, + set `config.api_key` to it, rebuild via `_setup_client()`. + 4. If no good key remains → raise `KeyPoolExhaustedError` with a masked + per-key summary (`sk-…a1b2: out of credit`) and an "add one with `/key add`" + hint. + 5. Otherwise fire `on_notice(...)` and retry. +- **`_blocking_chat`** runs inside `_with_failover` directly (the call is atomic). +- **Streaming boundary (`_stream_chat`):** auto-retry only if the failure happens + **before the first token** is emitted (exactly when auth/quota errors fire). If + any token already streamed, re-raise rather than duplicating output: + + ```python + emitted = False + try: + stream = self.client.chat.completions.create(..., stream=True) + for chunk in stream: + tok = ... + if tok: + emitted = True + on_token(tok) + except Exception as e: + if emitted or _key_failure_reason(e) is None or not failover_enabled: + raise + # rotate to next key (or raise KeyPoolExhaustedError) and retry from the top + ``` + +- **Notice plumbing (UI-agnostic):** `AIEngine` gains an + `on_notice: Callable[[str], None] | None` attribute (no signature change to + modes). The CLI sets it once to `print_warning` via a `_wire_engine()` helper, + re-applied wherever the engine is rebuilt (`/key`, `/model`, `/provider`). The + GUI may set its own sink. Message form: + `DeepSeek key #1 out of credit — switched to key #2`. +- **`KeyPoolExhaustedError`** is a `RuntimeError` subclass so existing CLI/agent + error handling surfaces it normally, while tests can assert on its type. + +### 3. REPL key-management commands (`cli.py`) + +CLI handlers are thin wrappers over the `config.py` pool helpers; after any +mutation they call `save_config()` and re-wire the engine. + +- **`/key `** — unchanged entry point. Sets the active key; for DeepSeek the + key is moved/inserted to the front of the pool. A real key is `sk-…`, so it + never collides with the subcommand words below. Routing lives in `_set_key`: + if the first token is `add` / `remove` / `list`, dispatch to the pool handler; + otherwise treat the whole argument as a key (today's behavior). +- **`/key add `** — validate via `validate_api_key`, then append to the pool + (dedup). Warn-but-add on failed validation (consistent with today's `_set_key`, + which saves even when validation fails). +- **`/key remove `** — remove pool entry `#n` (1-indexed). If it was active, + the new active becomes `pool[0]` and the client rebuilds. Out-of-range or + non-numeric → usage error. +- **`/keys`** — new entry in the command dispatch table; lists the pool masked + (`sk-…a1b2`), marking the active key (`●`) and any session-bad keys + (`✗ `), plus the count. `/key list` aliases to it. +- **Scope guard:** pool commands are DeepSeek-only. On another provider, + `/key add` prints `Multi-key pooling is DeepSeek-only — switch with /provider + deepseek`; `/keys` notes the same and shows the single active key. `/key + ` works on every provider. + +### 4. Error handling & edge cases + +- **All keys exhausted mid-session** → `KeyPoolExhaustedError` propagates through + `_handle_message` / the agent loop, printing the masked summary + `/key add` hint. +- **Removing the last key** → empty pool → falls back to today's + "API key not configured" messaging. +- **Duplicate add** → no-op with a note. +- **Adding an invalid key** → warn but still add (mirrors current behavior so a + transient validation failure does not block the user). +- **Non-DeepSeek provider** → pool is ignored by the engine; single-attempt + behavior, byte-for-byte unchanged. +- The rotation notice prints before the response streams (failover happens before + the first token), so it never interleaves with streamed output. + +## Testing + +- **`tests/test_engine.py`** + - `_key_failure_reason` table: `401`/`403`/`402`/`429+insufficient_quota` → + rotate reasons; plain `429`, `404`, connection → `None`. + - Failover rotates: mocked client raises `402` on key 1, succeeds on key 2 → + asserts retry happened, notice fired, key 2's content returned. + - All-exhausted → `KeyPoolExhaustedError` with masked summary. + - **Non-DeepSeek provider:** `402` raises immediately, no rotation. + - Streaming: failure before first token rotates and retries; failure after a + token re-raises (no duplicate output). + - All via a mocked OpenAI client — no network calls. +- **`tests/test_config.py`** + - Reconciliation: single key → pool seeded; pool → `api_key == pool[0]`; + dedupe + active-first; env key prepended. + - `api_keys` save/load round-trip. + - `add_key` / `remove_key` / `set_active_key` helpers (including invariant and + bad-index handling). + +## Backward compatibility + +- Existing `config.yaml` files with a single `ai.api_key` load unchanged and are + silently upgraded to a one-entry pool on first load. +- `config.ai.api_key` keeps its meaning (the active key) for every existing reader. +- `/key `, `hackbot setup`, and all provider env vars keep working as before. + +## Future work (explicitly deferred) + +- GUI management UI for the pool (`/api/config` could accept `api_keys`). +- Optional comma-separated bulk entry (`DEEPSEEK_API_KEY=k1,k2,k3`, `setup`). +- Generalizing the pool to other providers, if ever wanted. diff --git a/docs/superpowers/specs/2026-06-22-autonomous-tool-install-design.md b/docs/superpowers/specs/2026-06-22-autonomous-tool-install-design.md new file mode 100644 index 0000000..f53f552 --- /dev/null +++ b/docs/superpowers/specs/2026-06-22-autonomous-tool-install-design.md @@ -0,0 +1,214 @@ +# Design: Autonomous Security-Tool Installation + +Date: 2026-06-22 +Status: Approved (pending spec review) + +## Problem + +HackBot drives real security tools (nmap, nikto, sqlmap, nuclei, …) during an +autonomous assessment. When a tool the agent wants to use is not installed, the +runner returns `Tool not found` and the agent is stuck — the user must leave the +session, install the tool by hand, and start over. We want HackBot to install +missing security tools itself. + +## Goals + +- The agent can autonomously install a missing tool mid-assessment via a new + `install` action, then continue using it. +- Installation reuses HackBot's existing, deliberately hardened execution and + privilege boundary rather than introducing a new one. +- Cross-distro / cross-manager reality is handled: many security tools ship via + `go install` or `pipx`, not just `apt`. +- The capability is bounded by default (allowlist-only) and respects the user's + existing `safe_mode` trust dial. + +## Non-Goals + +- Removing or installing arbitrary system packages unrelated to security tooling + (only allowed when the user explicitly opts in — see `allow_arbitrary_install`). +- Self-updating HackBot itself (that already exists in `core/updater.py`). +- Managing tool *versions* / upgrades. First version installs if missing; it does + not upgrade an already-present tool. + +## Decisions (from brainstorming) + +| Question | Decision | +| --- | --- | +| Trigger | Fully autonomous — new `install` agent action. Plus a manual `/install` + `hackbot install` surface for testing/users. | +| What can be installed | Allowlist-bounded (tools already in `agent.allowed_tools`) with a curated tool→package map. Opt-in `allow_arbitrary_install` relaxes this. | +| Managers / platforms | Layered: system manager (`apt-get`/`dnf`/`pacman`/`brew`) + language managers (`pipx`/`pip`, `go install`), pluggable. | +| Privilege | Reuse centralized `_apply_sudo` in `ToolRunner`. No new privilege path. | +| Confirmation | Respect `safe_mode`: on (default) → one y/n confirmation per install; `--no-safe-mode` → unattended. | +| `allow_arbitrary_install` default | Off (allowlist-only). | + +## Architecture + +### 1. `hackbot/core/installer.py` — `ToolInstaller` + +A self-contained engine, structured like other `core/` modules. + +- **Manager detection (once, cached):** probe `apt-get`, `dnf`, `pacman`, `brew`, + `pipx`, `pip`, `go` via `shutil.which`. Expose `available_managers()`. +- **`plan_install(tool) -> Optional[InstallPlan]`:** consult `TOOL_INSTALL_MAP`, + pick the first manager in the tool's preferred order that is available on this + system, and build an `InstallPlan`. Returns `None` if the tool is unmappable or + no suitable manager is present (caller surfaces a clear reason). +- **`install(tool, runner) -> InstallResult`:** execute the plan's argument vector + through the supplied `ToolRunner` (NOT a fresh subprocess), then re-check + `resolve_tool_path(tool)` to confirm the binary now exists. Returns success + + resolved path, or failure + captured output. + +**`InstallPlan` (dataclass):** `tool`, `manager`, `package`, `command` (list[str], +already split — no shell), `needs_sudo: bool`. + +**`InstallResult` (dataclass):** `tool`, `success: bool`, `manager`, `path: +Optional[str]`, `message`, `stdout`/`stderr` (truncated). + +**Dependencies:** `ToolRunner`, `resolve_tool_path`, `TOOL_INSTALL_MAP`. One job: +turn "I need tool X" into a verified install or a clear, structured failure. + +### 2. `TOOL_INSTALL_MAP` — in `config.py` (beside `TOOL_ALIASES`) + +```python +TOOL_INSTALL_MAP = { + "nuclei": {"order": ["go", "apt"], + "go": "github.com/projectdiscovery/nuclei/v3/cmd/nuclei@latest", + "apt": "nuclei"}, + "subfinder": {"order": ["go"], + "go": "github.com/projectdiscovery/subfinder/v2/cmd/subfinder@latest"}, + "sqlmap": {"order": ["apt", "pipx"], "apt": "sqlmap", "pipx": "sqlmap"}, + # ...one entry per supported tool... +} +``` + +- `order` lists preferred managers; the installer picks the first one available. +- Per-manager values are the package/module identifier for that manager. +- `apt` entries are also used for `dnf`/`pacman` when the package name matches; + where names differ, the map carries `dnf`/`pacman` keys explicitly. `brew` and + `pip` are keyed the same way. +- Only tools present in `agent.allowed_tools` are installable. The initial map + covers the commonly-missing tools first; it does not need an entry for every one + of the ~80 allowlisted tools on day one, but unmapped tools simply report + "no install recipe" rather than failing silently. + +### 3. New agent action — `install` + +Format (documented in `engine.py`'s agent system prompt alongside the other +actions): + +```json +{"action": "install", "tool": "nuclei", "explanation": "needed to scan for known CVEs"} +``` + +Dispatch flow added to `agent.py` (`_parse_actions` recognizes it; +`_process_actions_loop` handles it): + +1. **Allowlist check** — reject if `tool` not in `allowed_tools`, unless + `allow_arbitrary_install` is set. Rejection text is fed back to the AI. +2. **Already installed?** — if `resolve_tool_path(tool)` already resolves, report + "already installed" back to the AI (no-op) and continue. +3. **Plan** — `installer.plan_install(tool)`. If `None`, feed "can't install, + here's why (unmapped / no manager)" back to the AI so it can adapt. +4. **Confirmation gate** — if `safe_mode` is on, pause for one y/n confirmation + using the existing confirmation UI, showing the exact command. If declined, + feed the decline back to the AI. With `--no-safe-mode`, proceed unattended. +5. **Execute** — `installer.install(...)`; feed the structured result (success + + new path, or failure output) back into the loop so the agent retries the tool + or adapts. + +The existing loop guards (dedupe, repeat caps, repeated-failure stop) apply to +`install` actions too, preventing install retry storms. + +The `engine.py` prompt is updated to tell the agent that when a tool run returns +`Tool not found`, it may emit an `install` action for that tool. No separate +runner hook is required — the "Tool not found" string is already fed back to the +AI. + +### 4. Security integration (preserve the hardened boundary) + +- Installs execute through `ToolRunner`, so `validate_command`, `BLOCKED_COMMANDS`, + and shell-operator rejection remain in force. +- Package managers used as install drivers (`apt-get`, `dnf`, `pacman`, `brew`, + `pipx`, `pip`, `go`) are **NOT** in the global `agent.allowed_tools` list. + `ToolInstaller.install` runs them via a dedicated `allow_install_drivers` + bypass on `ToolRunner.validate_command`/`execute` that permits only the + binaries listed in `config.INSTALL_DRIVERS`, so a normal `execute` action + from the agent cannot invoke them. Their argument vectors are constructed by + the installer from the curated map — never from raw LLM text — except in + opt-in `allow_arbitrary_install` mode, where the AI-named package string is + still `shlex`-safe-validated and passed as a single argument (no shell). +- Sudo handling stays centralized in `ToolRunner._apply_sudo`. The installer sets + `needs_sudo` on the plan; the runner remains the single sudo authority. +- `RISKY_PATTERNS` / `safe_mode` confirmation behavior is reused, not bypassed. + +### 5. Config + +Add to the `agent` config section (`config.py`): + +- `allow_arbitrary_install: bool = False` — when true, the agent may install a + package the LLM names even if it is not in the curated map / allowlist. + +No other config changes. `safe_mode` and `sudo_*` already exist and are reused. + +### 6. Manual surface (reuses the installer module) + +- **REPL:** `/install ` slash command → `_install_tool` method on + `HackBotApp`, registered in the command table. Plans, confirms (respecting + `safe_mode`), installs, prints result. +- **CLI:** `hackbot install ` subcommand in the Click group, same path. + +These exist because the installer is a clean module and this makes the feature +testable and directly usable; they add negligible surface. + +## Data Flow + +``` +agent LLM emits {"action":"install","tool":"nuclei"} + -> agent._process_actions_loop + -> allowlist + already-installed checks + -> ToolInstaller.plan_install("nuclei") -> InstallPlan(go install ...) + -> safe_mode? confirm y/n + -> ToolInstaller.install(plan, runner) + -> ToolRunner.execute(["go","install",...]) (shlex, sudo if needed) + -> resolve_tool_path("nuclei") to verify + -> InstallResult fed back into conversation + -> agent re-runs the original tool command +``` + +## Error Handling + +- **Unmapped tool / no available manager:** structured "no install recipe" + message back to the AI; agent adapts (different tool or `complete`). +- **Install command fails (nonzero exit):** captured stdout/stderr (truncated) + returned; loop guards stop repeated identical failures. +- **Binary still missing after "successful" install:** treated as failure + (re-check via `resolve_tool_path`), reported as such. +- **User declines confirmation:** decline fed back; no install. +- **No package managers at all:** `available_managers()` empty → every plan is + `None` → clear message. + +## Testing — `tests/test_installer.py` + +All tests mock `shutil.which` and the `ToolRunner`; **no real packages installed.** + +- Manager detection with various `which` availabilities. +- `plan_install`: correct manager chosen by `order`; correct command vector; + `needs_sudo` set correctly; `None` for unmapped tool; `None` when no manager. +- Allowlist enforcement: install of non-allowlisted tool rejected unless + `allow_arbitrary_install`. +- `safe_mode` gate: confirmation requested when on; skipped when off (mock the + confirmation callback). +- Install round-trip with a mocked runner: success path (binary "appears") and + failure path (nonzero exit / binary still missing). +- Agent dispatch: `_parse_actions` recognizes the `install` action; already- + installed short-circuit. + +## Files Touched + +- `hackbot/core/installer.py` — new module. +- `hackbot/config.py` — `TOOL_INSTALL_MAP`, `allow_arbitrary_install` flag, + add managers to allowlist as install drivers. +- `hackbot/modes/agent.py` — parse + dispatch `install` action. +- `hackbot/core/engine.py` — document the `install` action in the agent prompt. +- `hackbot/cli.py` — `/install` slash command + `hackbot install` subcommand. +- `tests/test_installer.py` — new tests. diff --git a/hackbot/cli.py b/hackbot/cli.py index c0c720f..1e47f74 100644 --- a/hackbot/cli.py +++ b/hackbot/cli.py @@ -74,6 +74,12 @@ load_dotenv() +PROVIDER_ALIASES = { + "deep": "deepseek", + "deepseek": "deepseek", +} + + def _should_block_root_gui_launch() -> bool: """Return True if GUI launch should be blocked due to root/sudo context. @@ -134,6 +140,14 @@ def _on_agent_step(self, step) -> None: elif step.action == "report": print_success(step.description) + def _rebuild_ai_engine(self) -> None: + self.engine = AIEngine(self.config.ai) + self.chat.engine = self.engine + self.plan.engine = self.engine + if self.agent: + self.agent.engine = self.engine + self.agent.summarizer.engine = self.engine + def _on_confirm(self, command: str, reason: str) -> bool: """Handle confirmation for risky commands.""" return confirm_action(command, reason) @@ -177,6 +191,7 @@ def _handle_command(self, text: str) -> bool: "/plan": lambda: self._switch_mode("plan"), "/config": lambda: self._show_config(), "/tools": lambda: self._show_tools(), + "/install": lambda: self._install_tool(args), "/model": lambda: self._set_model(args), "/key": lambda: self._set_key(args), "/provider": lambda: self._set_provider(args), @@ -399,6 +414,23 @@ def _show_tools(self) -> bool: show_tools_status(tools) return True + def _install_tool(self, args: str) -> bool: + tool = args.strip() + if not tool: + print_error("Usage: /install ") + return True + if self.mode != "agent" or not self.agent: + self.agent = AgentMode( + engine=self.engine, + config=self.config, + on_step=self._on_agent_step, + on_confirm=self._on_confirm, + on_output=self._on_tool_output, + ) + result = self.agent._process_install_action({"action": "install", "tool": tool}) + console.print(Markdown(result)) + return True + def _set_model(self, model: str) -> bool: if not model: print_error("Usage: /model \n See available models: /models") @@ -492,15 +524,35 @@ def _check_update(self, args: str) -> bool: def _set_key(self, key: str) -> bool: if not key: - print_error("Usage: /key ") + print_error("Usage: /key or /key deepseek ") + return True + + key_parts = key.split(maxsplit=1) + requested_provider = PROVIDER_ALIASES.get(key_parts[0].lower()) + if requested_provider: + if len(key_parts) == 1: + print_error(f"Usage: /key {key_parts[0]} ") + print_info(f"Or run: /provider {requested_provider} then /key ") + return True + + preset = PROVIDERS[requested_provider] + self.config.ai.provider = requested_provider + if preset["models"]: + self.config.ai.model = preset["models"][0]["id"] + self.config.ai.base_url = "" + key = key_parts[1].strip() + if not key: + print_error(f"Usage: /key {key_parts[0]} ") + return True + print_info(f"Provider: {preset['name']}") + + if key.lower() in PROVIDER_ALIASES: + print_error(f"Usage: /key {key} ") + print_info(f"Or run: /provider {PROVIDER_ALIASES[key.lower()]} then /key ") return True + self.config.ai.api_key = key - self.engine = AIEngine(self.config.ai) - self.chat.engine = self.engine - self.plan.engine = self.engine - if self.agent: - self.agent.engine = self.engine - self.agent.summarizer.engine = self.engine + self._rebuild_ai_engine() # Validate the key before saving print_info("Validating API key...") @@ -730,12 +782,7 @@ def _set_provider(self, provider: str) -> bool: self.config.ai.model = preset["models"][0]["id"] # Clear base_url so engine uses preset self.config.ai.base_url = "" - self.engine = AIEngine(self.config.ai) - self.chat.engine = self.engine - self.plan.engine = self.engine - if self.agent: - self.agent.engine = self.engine - self.agent.summarizer.engine = self.engine + self._rebuild_ai_engine() save_config(self.config) print_success( f"Provider: {preset['name']}\n" @@ -2607,6 +2654,16 @@ def tools(ctx): show_tools_status(tool_status) +@main.command(name="install") +@click.argument("tool") +@click.pass_context +def install_cmd(ctx, tool): + """Install a security tool that isn't present.""" + config = ctx.obj["config"] + app = HackBotApp(config) + app._install_tool(tool) + + @main.command(name="update") @click.option("--force", is_flag=True, help="Force reinstall even if up to date") @click.option("--check", is_flag=True, help="Only check for updates, don't install") diff --git a/hackbot/config.py b/hackbot/config.py index 7024a48..f542ca9 100644 --- a/hackbot/config.py +++ b/hackbot/config.py @@ -41,6 +41,7 @@ def ensure_dirs() -> None: "provider": "openai", "model": "gpt-4o", "api_key": "", + "api_keys": [], "base_url": "", "temperature": 0.2, "max_tokens": 4096, @@ -53,6 +54,7 @@ def ensure_dirs() -> None: "sudo_mode": False, "sudo_password": "", "nvd_api_key": "", + "allow_arbitrary_install": False, "allowed_tools": [ "nmap", "masscan", @@ -165,6 +167,7 @@ class AIConfig: provider: str = "openai" model: str = "gpt-4o" api_key: str = "" + api_keys: List[str] = field(default_factory=list) base_url: str = "" temperature: float = 0.2 max_tokens: int = 4096 @@ -179,7 +182,10 @@ class AgentConfig: sudo_mode: bool = False sudo_password: str = "" nvd_api_key: str = "" - allowed_tools: List[str] = field(default_factory=lambda: DEFAULT_CONFIG["agent"]["allowed_tools"]) + allow_arbitrary_install: bool = False + allowed_tools: List[str] = field( + default_factory=lambda: DEFAULT_CONFIG["agent"]["allowed_tools"] + ) @dataclass @@ -222,6 +228,62 @@ class HackBotConfig: rag: RAGConfig = field(default_factory=RAGConfig) +def reconcile_keys(ai: Dict[str, Any]) -> None: + """Normalize the API key pool in an ``ai`` config dict, in place. + + Dedupes, drops empties, and keeps the active key first so the invariant + ``api_key == api_keys[0]`` always holds. Seeds the pool from a legacy + single ``api_key`` and lets an env-provided ``api_key`` take priority by + placing it first. + """ + pool = [k for k in dict.fromkeys([ai.get("api_key", "")] + list(ai.get("api_keys") or [])) if k] + ai["api_keys"] = pool + ai["api_key"] = pool[0] if pool else "" + + +def add_key(cfg: "AIConfig", key: str) -> None: + """Append *key* to the pool if not already present; keep ``api_key`` valid.""" + key = key.strip() + if not key or key in cfg.api_keys: + return + cfg.api_keys.append(key) + if not cfg.api_key: + cfg.api_key = cfg.api_keys[0] + + +def remove_key(cfg: "AIConfig", index: int) -> str: + """Remove and return the pool entry at *index*. Raises IndexError if invalid. + + If the removed key was the active one, the new active key becomes the new + ``api_keys[0]`` (or empty when the pool is exhausted). + """ + if index < 0 or index >= len(cfg.api_keys): + raise IndexError("API key index out of range") + removed = cfg.api_keys.pop(index) + cfg.api_key = cfg.api_keys[0] if cfg.api_keys else "" + return removed + + +def set_active_key(cfg: "AIConfig", key: str) -> None: + """Make *key* the active key by moving/inserting it at the front of the pool.""" + key = key.strip() + if not key: + return + if key in cfg.api_keys: + cfg.api_keys.remove(key) + cfg.api_keys.insert(0, key) + cfg.api_key = key + + +def mask_key(key: str) -> str: + """Return a display-safe masked form of an API key.""" + if not key: + return "(empty)" + if len(key) <= 8: + return key[:3] + "…" + return f"{key[:5]}…{key[-4:]}" + + def load_config() -> HackBotConfig: """Load configuration from disk, env vars, and defaults.""" ensure_dirs() @@ -305,6 +367,8 @@ def load_config() -> HackBotConfig: DEFAULT_CONFIG["agent"]["allowed_tools"], ) + reconcile_keys(merged.setdefault("ai", {})) + cfg = HackBotConfig( ai=AIConfig(**merged.get("ai", {})), agent=AgentConfig(**merged.get("agent", {})), @@ -324,6 +388,7 @@ def save_config(cfg: HackBotConfig) -> None: "provider": cfg.ai.provider, "model": cfg.ai.model, "api_key": cfg.ai.api_key, + "api_keys": cfg.ai.api_keys, "base_url": cfg.ai.base_url, "temperature": cfg.ai.temperature, "max_tokens": cfg.ai.max_tokens, @@ -336,6 +401,7 @@ def save_config(cfg: HackBotConfig) -> None: "sudo_mode": cfg.agent.sudo_mode, "sudo_password": cfg.agent.sudo_password, "nvd_api_key": cfg.agent.nvd_api_key, + "allow_arbitrary_install": cfg.agent.allow_arbitrary_install, "allowed_tools": cfg.agent.allowed_tools, }, "reporting": { @@ -404,6 +470,7 @@ def _merge_allowed_tools(current: List[str], defaults: List[str]) -> List[str]: # ── tool detection ─────────────────────────────────────────────────────────── + def detect_platform() -> Dict[str, str]: """Return platform information.""" return { @@ -445,6 +512,71 @@ def detect_platform() -> Dict[str, str]: ], } +# Package-manager binaries the ToolInstaller is permitted to invoke. These are +# deliberately NOT in agent.allowed_tools so the agent cannot run them via a +# normal `execute` action — only ToolInstaller.install() may, via the runner's +# allow_install_drivers bypass. +INSTALL_DRIVERS: List[str] = ["apt-get", "dnf", "pacman", "brew", "pipx", "pip", "go"] + +# Curated tool -> package recipes for autonomous installation. +# `order` lists preferred managers; ToolInstaller picks the first one available +# on the host. Per-manager values are the package/module identifier for that +# manager. Only tools also present in agent.allowed_tools are installable. +TOOL_INSTALL_MAP: Dict[str, Dict[str, object]] = { + "nuclei": { + "order": ["go", "apt"], + "go": "github.com/projectdiscovery/nuclei/v3/cmd/nuclei@latest", + "apt": "nuclei", + }, + "subfinder": { + "order": ["go"], + "go": "github.com/projectdiscovery/subfinder/v2/cmd/subfinder@latest", + }, + "httpx": { + "order": ["go"], + "go": "github.com/projectdiscovery/httpx/cmd/httpx@latest", + }, + "ffuf": { + "order": ["go", "apt"], + "go": "github.com/ffuf/ffuf/v2@latest", + "apt": "ffuf", + }, + "gobuster": { + "order": ["go", "apt"], + "go": "github.com/OJ/gobuster/v3@latest", + "apt": "gobuster", + }, + "katana": { + "order": ["go"], + "go": "github.com/projectdiscovery/katana/cmd/katana@latest", + }, + "sqlmap": {"order": ["apt", "pipx"], "apt": "sqlmap", "pipx": "sqlmap"}, + "nikto": {"order": ["apt"], "apt": "nikto"}, + "nmap": { + "order": ["apt", "dnf", "pacman", "brew"], + "apt": "nmap", + "dnf": "nmap", + "pacman": "nmap", + "brew": "nmap", + }, + "whatweb": {"order": ["apt"], "apt": "whatweb"}, + "wpscan": {"order": ["apt"], "apt": "wpscan"}, + "hydra": {"order": ["apt"], "apt": "hydra"}, + "dnsrecon": {"order": ["apt", "pipx"], "apt": "dnsrecon", "pipx": "dnsrecon"}, + "amass": { + "order": ["apt", "go"], + "apt": "amass", + "go": "github.com/owasp-amass/amass/v4/...@master", + }, + "feroxbuster": {"order": ["apt"], "apt": "feroxbuster"}, + "masscan": { + "order": ["apt", "dnf", "pacman"], + "apt": "masscan", + "dnf": "masscan", + "pacman": "masscan", + }, +} + def resolve_tool_path(tool: str) -> Optional[str]: """Resolve a logical tool name to an installed executable path. diff --git a/hackbot/core/engine.py b/hackbot/core/engine.py index b840edd..e6011c4 100644 --- a/hackbot/core/engine.py +++ b/hackbot/core/engine.py @@ -282,6 +282,16 @@ Sends N simultaneous requests using thread barriers and analyzes response differences to detect race conditions. +10. **Autonomous Tool Installation** — If a tool you need is not installed (a command + returns "Tool not found"), you can install it autonomously: + ```json + {"action": "install", "tool": "", "explanation": ""} + ``` + Only allowlisted security tools with a known install recipe can be installed. + When safe_mode is on, the user confirms each install. After a successful install, + re-run the original command. If the install fails or no recipe exists, adapt and + use a different tool. + ZERO-DAY METHODOLOGY: - After initial recon, identify services with unusual behaviors or outdated versions - Use nuclei with community templates for latest vulnerability checks diff --git a/hackbot/core/installer.py b/hackbot/core/installer.py new file mode 100644 index 0000000..62ed7e7 --- /dev/null +++ b/hackbot/core/installer.py @@ -0,0 +1,126 @@ +"""Autonomous installation of missing security tools. + +ToolInstaller turns a logical tool name into a verified install by choosing an +available package manager from config.TOOL_INSTALL_MAP and executing the +constructed argv through the existing ToolRunner (shell-free, sudo-aware). +""" + +from __future__ import annotations + +import shutil +from dataclasses import dataclass + +from hackbot.config import TOOL_INSTALL_MAP, resolve_tool_path + +# Logical manager key -> binary probed via shutil.which. +_MANAGER_BINARY: dict[str, str] = { + "apt": "apt-get", + "dnf": "dnf", + "pacman": "pacman", + "brew": "brew", + "pipx": "pipx", + "pip": "pip", + "go": "go", +} + +# System managers that require root for a system-wide install. +_SUDO_MANAGERS = {"apt", "dnf", "pacman"} + + +@dataclass +class InstallPlan: + tool: str + manager: str + package: str + command: list[str] + needs_sudo: bool + + +@dataclass +class InstallResult: + tool: str + success: bool + manager: str = "" + path: str | None = None + message: str = "" + stdout: str = "" + stderr: str = "" + + +def _build_command(manager: str, package: str) -> list[str]: + if manager == "apt": + return ["apt-get", "install", "-y", package] + if manager == "dnf": + return ["dnf", "install", "-y", package] + if manager == "pacman": + return ["pacman", "-S", "--noconfirm", package] + if manager == "brew": + return ["brew", "install", package] + if manager == "pipx": + return ["pipx", "install", package] + if manager == "pip": + return ["pip", "install", package] + if manager == "go": + return ["go", "install", package] + raise ValueError(f"unknown manager: {manager}") + + +class ToolInstaller: + def __init__(self, runner, install_map: dict | None = None): + self.runner = runner + self.install_map = install_map if install_map is not None else TOOL_INSTALL_MAP + self._available: dict[str, str] | None = None + + def available_managers(self) -> dict[str, str]: + if self._available is None: + found: dict[str, str] = {} + for mgr, binary in _MANAGER_BINARY.items(): + path = shutil.which(binary) + if path: + found[mgr] = path + self._available = found + return self._available + + def plan_install(self, tool: str) -> InstallPlan | None: + recipe = self.install_map.get(tool.lower()) + if not recipe: + return None + available = self.available_managers() + order = recipe.get("order") or [k for k in recipe if k != "order"] + for mgr in order: + package = recipe.get(mgr) + if package and mgr in available: + return InstallPlan( + tool=tool, + manager=mgr, + package=str(package), + command=_build_command(mgr, str(package)), + needs_sudo=mgr in _SUDO_MANAGERS, + ) + return None + + def install(self, plan: InstallPlan) -> InstallResult: + command_str = " ".join(plan.command) + result = self.runner.execute( + command_str, + tool_name=plan.manager, + explanation=f"Install {plan.tool}", + allow_install_drivers=True, + ) + path = resolve_tool_path(plan.tool) + success = result.success and path is not None + if success: + message = f"Installed {plan.tool} via {plan.manager} ({path})" + elif result.success: + message = f"{plan.manager} reported success but {plan.tool} is still not on PATH" + else: + message = f"Install of {plan.tool} via {plan.manager} failed" + return InstallResult( + tool=plan.tool, + success=success, + manager=plan.manager, + path=path, + message=message, + stdout=(result.stdout or "")[:2000], + stderr=(result.stderr or "")[:2000], + ) diff --git a/hackbot/core/runner.py b/hackbot/core/runner.py index bc35314..c4ea503 100644 --- a/hackbot/core/runner.py +++ b/hackbot/core/runner.py @@ -19,7 +19,7 @@ from pathlib import Path from typing import Any, Callable, Dict, List, Optional -from hackbot.config import LOGS_DIR, resolve_tool_path +from hackbot.config import INSTALL_DRIVERS, LOGS_DIR, resolve_tool_path # Lazy reference — filled at runtime to avoid circular imports _plugin_manager = None @@ -108,6 +108,11 @@ def output(self) -> str: "bash -i", ] +# Standalone shell-operator tokens that indicate command chaining or redirection. +# Detected as discrete tokens (via shlex), so characters inside a single argument +# (e.g. an '&' in a URL query string) are never flagged. +SHELL_OPERATOR_TOKENS = frozenset({";", "|", "||", "&", "&&", ">", ">>", "<", "<<"}) + class ToolRunner: """ @@ -313,7 +318,56 @@ def _extract_validated_tool(self, parts: List[str]) -> str: return candidate - def validate_command(self, command: str) -> tuple[bool, str]: + @staticmethod + def _strip_wrapping_quotes(value: str) -> str: + """Remove one layer of matching shell quotes from a token.""" + if len(value) >= 2 and value[0] == value[-1] and value[0] in {"'", '"'}: + return value[1:-1] + return value + + @staticmethod + def _split_command(command: str) -> List[str]: + """Split command text for validation without invoking a shell.""" + if platform.system() == "Windows": + return [ + ToolRunner._strip_wrapping_quotes(part) + for part in shlex.split(command, posix=False) + ] + return shlex.split(command) + + @staticmethod + def _contains_shell_operators(command: str) -> Optional[str]: + """Return the offending marker if *command* contains shell command + substitution or a standalone shell-operator token, else None. + + Operates on the NORMALIZED command (wrapping backticks and a leading + ``$ `` prompt have already been stripped by ``_normalize_command``), so + only *embedded* substitution survives to be rejected here. Detection is + token-based to avoid false positives on legitimate arguments such as URL + query strings (``http://x/?a=1&b=2`` stays inside a single shlex token). + + Returns ``""`` when the command cannot be tokenized + (e.g. unbalanced quotes) so the caller can reject it gracefully. + """ + # Command substitution: any remaining backtick or "$(" is rejected. + if "`" in command: + return "`" + if "$(" in command: + return "$(" + + try: + tokens = ToolRunner._split_command(command) + except ValueError: + return "" + + for tok in tokens: + if tok in SHELL_OPERATOR_TOKENS: + return tok + return None + + def validate_command( + self, command: str, allow_install_drivers: bool = False, + ) -> tuple[bool, str]: """ Validate a command for safety. Returns (is_safe, reason). @@ -330,8 +384,21 @@ def validate_command(self, command: str) -> tuple[bool, str]: if blocked in cmd_lower: return False, f"Blocked command detected: {blocked}" + # Reject shell command-substitution and standalone operator tokens. + # Unconditional (not safe_mode-gated): execution is shell-free, so this + # is a clear, early hard-reject of unsupported syntax rather than a + # confirmable RISKY warning. + offender = self._contains_shell_operators(command) + if offender == "": + return False, "Invalid command: unbalanced quotes" + if offender is not None: + return False, f"Shell metacharacter not allowed: '{offender}'" + # Extract tool name (skip 'sudo' prefix for validation) - parts = shlex.split(command) if platform.system() != "Windows" else command.split() + try: + parts = self._split_command(command) + except ValueError: + return False, "Invalid command: unbalanced quotes" if not parts: return False, "Empty command" @@ -341,7 +408,9 @@ def validate_command(self, command: str) -> tuple[bool, str]: # Check if tool is allowed if not self.is_tool_allowed(tool): - return False, f"Tool '{tool}' is not in the allowed list" + driver_ok = allow_install_drivers and os.path.basename(tool).lower() in INSTALL_DRIVERS + if not driver_ok: + return False, f"Tool '{tool}' is not in the allowed list" # Check risky patterns in safe mode if self.safe_mode: @@ -381,12 +450,15 @@ def check_sudo(self) -> tuple[bool, str]: Returns (ok, message). Sets ``_sudo_validated`` on success so the check is only performed once per runner lifetime. """ - if not self.sudo_mode or platform.system() == "Windows": + if not self.sudo_mode: return True, "sudo not required" if self._sudo_validated: return True, "sudo already validated" + if platform.system() == "Windows": + return True, "sudo not required" + try: if self.sudo_password: proc = subprocess.Popen( @@ -423,7 +495,10 @@ def check_sudo(self) -> tuple[bool, str]: except Exception as e: return False, f"sudo check error: {e}" - def execute(self, command: str, tool_name: str = "", explanation: str = "") -> ToolResult: + def execute( + self, command: str, tool_name: str = "", explanation: str = "", + allow_install_drivers: bool = False, + ) -> ToolResult: """ Execute a command synchronously with timeout and output capture. """ @@ -437,7 +512,9 @@ def execute(self, command: str, tool_name: str = "", explanation: str = "") -> T command = self._apply_sudo(command) # Validate - is_safe, reason = self.validate_command(command) + is_safe, reason = self.validate_command( + command, allow_install_drivers=allow_install_drivers, + ) if not is_safe: return ToolResult( @@ -472,11 +549,13 @@ def execute(self, command: str, tool_name: str = "", explanation: str = "") -> T try: is_windows = platform.system() == "Windows" proc = subprocess.Popen( + # Windows: pass the raw string to CreateProcess (no cmd.exe, so + # shell metacharacters are NOT interpreted). POSIX: split to argv. command if is_windows else shlex.split(command), stdout=subprocess.PIPE, stderr=subprocess.PIPE, stdin=subprocess.PIPE if stdin_data else None, - shell=is_windows, + shell=False, text=True, env=self._get_env(), ) @@ -491,10 +570,7 @@ def execute(self, command: str, tool_name: str = "", explanation: str = "") -> T duration = time.time() - start # Truncate if needed - truncated = False - if len(stdout) > self.MAX_OUTPUT_SIZE: - stdout = stdout[: self.MAX_OUTPUT_SIZE] + f"\n\n[OUTPUT TRUNCATED at {self.MAX_OUTPUT_SIZE} bytes]" - truncated = True + stdout, truncated = self._truncate_output(stdout) result = ToolResult( tool=self._infer_tool_name(command, tool_name), @@ -540,12 +616,17 @@ def execute(self, command: str, tool_name: str = "", explanation: str = "") -> T return result async def execute_async(self, command: str, tool_name: str = "") -> ToolResult: - """Execute a command asynchronously.""" + """Execute a command asynchronously (shell-free; parity with execute()).""" command = self._normalize_command(command) + # Plugin execution — intercept hackbot-plugin commands + if command.strip().startswith("hackbot-plugin "): + return self._execute_plugin(command, tool_name) + # Apply sudo prefix if enabled command = self._apply_sudo(command) + # Validate is_safe, reason = self.validate_command(command) if not is_safe: return ToolResult( @@ -558,11 +639,46 @@ async def execute_async(self, command: str, tool_name: str = "") -> ToolResult: success=False, ) + # Check for risky commands + if "RISKY" in reason and not self.auto_confirm: + if self.on_confirm: + confirmed = self.on_confirm(command, reason) + if not confirmed: + return ToolResult( + tool=self._infer_tool_name(command, tool_name), + command=command, + stdout="", + stderr="User declined execution", + return_code=-2, + duration=0, + success=False, + ) + start = time.time() stdin_data = self._feed_sudo_password() if "sudo -S" in command else None + + # Tokenize without a shell so /bin/sh is never invoked (parity with execute()). try: - proc = await asyncio.create_subprocess_shell( - command, + args = shlex.split(command) + except ValueError as e: + result = ToolResult( + tool=self._infer_tool_name(command, tool_name), + command=command, + stdout="", + stderr=f"Execution error: {e}", + return_code=-4, + duration=time.time() - start, + success=False, + ) + self.history.append(result) + self._log_execution(result) + if self.on_output: + self.on_output(result.output) + return result + + try: + proc = await asyncio.create_subprocess_exec( + *args, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, stdin=asyncio.subprocess.PIPE if stdin_data else None, @@ -582,6 +698,9 @@ async def execute_async(self, command: str, tool_name: str = "") -> ToolResult: duration = time.time() - start + # Truncate if needed + stdout, truncated = self._truncate_output(stdout) + result = ToolResult( tool=self._infer_tool_name(command, tool_name), command=command, @@ -590,6 +709,18 @@ async def execute_async(self, command: str, tool_name: str = "") -> ToolResult: return_code=proc.returncode or 0, duration=duration, success=(proc.returncode or 0) == 0, + truncated=truncated, + ) + except FileNotFoundError: + duration = time.time() - start + result = ToolResult( + tool=self._infer_tool_name(command, tool_name), + command=command, + stdout="", + stderr=f"Tool not found: {self._infer_tool_name(command, tool_name)}", + return_code=-3, + duration=duration, + success=False, ) except Exception as e: duration = time.time() - start @@ -604,8 +735,18 @@ async def execute_async(self, command: str, tool_name: str = "") -> ToolResult: ) self.history.append(result) + self._log_execution(result) + if self.on_output: + self.on_output(result.output) return result + def _truncate_output(self, stdout: str) -> tuple[str, bool]: + """Cap *stdout* at MAX_OUTPUT_SIZE. Returns (text, truncated).""" + if len(stdout) <= self.MAX_OUTPUT_SIZE: + return stdout, False + notice = f"\n\n[OUTPUT TRUNCATED at {self.MAX_OUTPUT_SIZE} bytes]" + return stdout[: self.MAX_OUTPUT_SIZE] + notice, True + def _get_env(self) -> Dict[str, str]: """Get sanitized environment for subprocess execution.""" env = os.environ.copy() diff --git a/hackbot/gui/templates/index.html b/hackbot/gui/templates/index.html index 846970b..308519f 100644 --- a/hackbot/gui/templates/index.html +++ b/hackbot/gui/templates/index.html @@ -34,13 +34,23 @@ /* ── Layout ───────────────────────────────────────────────────────────────── */ .app-frame{display:flex;flex-direction:column;height:100vh;width:100vw} .app{display:flex;flex:1;overflow:hidden} -.sidebar{width:260px;min-width:260px;background:var(--bg2);border-right:1px solid var(--border);display:flex;flex-direction:column;overflow:hidden} +.sidebar{width:260px;min-width:260px;background:var(--bg2);border-right:1px solid var(--border);display:flex;flex-direction:column;overflow:hidden;transition:width .18s ease,min-width .18s ease} .main{flex:1;display:flex;flex-direction:column;overflow:hidden} +.app.sidebar-collapsed .sidebar{width:60px;min-width:60px} +.app.sidebar-collapsed .logo{padding:12px 8px;justify-content:center;flex-direction:column;gap:8px} +.app.sidebar-collapsed .logo-brand h1,.app.sidebar-collapsed .logo-brand span,.app.sidebar-collapsed .nav-section-title,.app.sidebar-collapsed .nav-btn span:not(.icon),.app.sidebar-collapsed .sidebar-footer{display:none} +.app.sidebar-collapsed .logo-brand img{margin-right:0!important} +.app.sidebar-collapsed .nav{padding:8px} +.app.sidebar-collapsed .nav-btn{justify-content:center;padding:12px} +.app.sidebar-collapsed .sidebar-toggle{width:32px;height:28px;justify-content:center} /* ── Sidebar ──────────────────────────────────────────────────────────────── */ -.logo{padding:20px;border-bottom:1px solid var(--border);text-align:center} -.logo h1{font-size:1.3rem;color:var(--green);letter-spacing:1px} -.logo span{font-size:.75rem;color:var(--text-dim)} +.logo{padding:16px 12px 16px 20px;border-bottom:1px solid var(--border);display:flex;align-items:center;justify-content:space-between;gap:10px} +.logo-brand{display:flex;align-items:center;gap:8px;min-width:0} +.logo h1{font-size:1.3rem;color:var(--green);letter-spacing:1px;white-space:nowrap} +.logo span{font-size:.75rem;color:var(--text-dim);white-space:nowrap} +.sidebar-toggle{width:30px;height:30px;border:1px solid var(--border);border-radius:6px;background:var(--bg3);color:var(--text);cursor:pointer;display:inline-flex;align-items:center;justify-content:center;font-size:1rem;line-height:1;transition:all .15s;flex-shrink:0} +.sidebar-toggle:hover{border-color:var(--border2);background:var(--bg4);color:var(--text-bright)} .nav{flex:1;padding:12px;overflow-y:auto} .nav-section{margin-bottom:16px} @@ -181,7 +191,10 @@ /* ── Responsive ───────────────────────────────────────────────────────────── */ @media(max-width:768px){ .sidebar{width:60px;min-width:60px} - .sidebar .logo h1,.sidebar .logo span,.nav-section-title,.nav-btn span:not(.icon),.sidebar-footer{display:none} + .logo{padding:12px 8px;justify-content:center;flex-direction:column;gap:8px} + .sidebar .logo-brand h1,.sidebar .logo-brand span,.nav-section-title,.nav-btn span:not(.icon),.sidebar-footer{display:none} + .sidebar .logo-brand img{margin-right:0!important} + .nav{padding:8px} .nav-btn{justify-content:center;padding:12px} .titlebar-title span.full{display:none} } @@ -213,17 +226,22 @@ -
+
-
+