From a75910a4e6c453ea4f5fc4f7d1323eabd49f0734 Mon Sep 17 00:00:00 2001 From: AIOSAI Date: Mon, 15 Jun 2026 23:45:46 -0700 Subject: [PATCH] =?UTF-8?q?fix(api,skills):=20harden=20secrets=20door=20?= =?UTF-8?q?=E2=80=94=20no=20secret=20value=20to=20stdout=20(DPLAN-0211,=20?= =?UTF-8?q?clears=20CodeQL=20#86-88)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit PR #640's only failing required check was Code scanning/CodeQL: 3 HIGH py/clear-text-logging-sensitive-data alerts where get-secret printed raw secret values to stdout. Research (OWASP, CodeQL rule source, secret-CLI survey) confirmed a real exposure — acute for AIPass since it runs inside Claude Code, which captures command stdout into model context, and the telegram skill shelled out to get-secret and parsed the token from stdout. @api (P1): - NEW apps/modules/secrets.py — in-process cross-branch door (get_secret, list_secrets) wrapping the auth handler; consumers import this, not the CLI. - get_secret_cmd rewritten: masked summary by default ('slug: set (N chars)'), --out FILE writes the raw value 0o600 and prints only the path, --list shows slug names via console.print. All 3 raw-value print() sinks removed. - bypass.json reasoning + README + help updated. @skills (P2): - telegram config._get_secret / list_bot_configs rewired from subprocess+stdout parse to the in-process aipass.api.apps.modules.secrets API; subprocess/json imports dropped. Tests + SKILL.md updated. Also: seedgo test_checkers_batch2.py — comment the synthetic sk-or-v1 fixture keys as FAKE (not real credentials). Verified: @api 504 tests + seedgo 100%; telegram 452/452; skills 252/252; no secret reaches stdout by any path. Co-Authored-By: Claude Opus 4.8 (1M context) --- src/aipass/api/.seedgo/bypass.json | 8 +- src/aipass/api/README.md | 12 +- src/aipass/api/apps/modules/api_key.py | 59 ++++-- src/aipass/api/apps/modules/secrets.py | 127 +++++++++++++ src/aipass/api/tests/test_secrets.py | 176 ++++++++++++++---- .../seedgo/tests/test_checkers_batch2.py | 3 + .../skills/.aipass/skills/telegram/SKILL.md | 2 +- .../skills/telegram/apps/handlers/config.py | 78 ++------ .../telegram/tests/test_multibot_config.py | 56 ++---- 9 files changed, 354 insertions(+), 167 deletions(-) create mode 100644 src/aipass/api/apps/modules/secrets.py diff --git a/src/aipass/api/.seedgo/bypass.json b/src/aipass/api/.seedgo/bypass.json index 22f7ced8..fdd0e6d3 100644 --- a/src/aipass/api/.seedgo/bypass.json +++ b/src/aipass/api/.seedgo/bypass.json @@ -133,13 +133,7 @@ { "file": "apps/modules/api_key.py", "standard": "cli", - "reason": "get_secret_cmd() uses bare print() intentionally — output must be machine-readable for agents that parse stdout from 'drone @api get-secret'. Rich formatting would break downstream consumers." - }, - { - "file": "apps/modules/api_key.py", - "standard": "debug_print", - "lines": [196, 208, 214], - "reason": "get_secret_cmd() uses bare print() intentionally — output must be machine-readable for agents that parse stdout from 'drone @api get-secret'. Rich formatting would break downstream consumers." + "reason": "get_secret_cmd() --list uses console.print() for slug names (identifiers, not secrets). Machine consumers use the in-process module aipass.api.apps.modules.secrets.get_secret; CLI never prints raw values (DPLAN-0211)." }, { "file": "tests/test_aggregation.py", diff --git a/src/aipass/api/README.md b/src/aipass/api/README.md index 697f646f..07fcd33a 100644 --- a/src/aipass/api/README.md +++ b/src/aipass/api/README.md @@ -5,7 +5,7 @@ > Centralized external API gateway — authenticated service clients for all external APIs **Module:** `aipass.api` | **Role:** `api_gateway` -**Seedgo:** 99% (36/37 at 100%) | **Tests:** 499 pass | **Functions:** 80 public (80 tested) +**Seedgo:** 100% (37/37 at 100%) | **Tests:** 504 pass | **Functions:** 82 public (82 tested) **Last Updated:** 2026-06-15 --- @@ -26,7 +26,7 @@ drone @api [args] | `validate [provider]` | Validate API key (default: openrouter) | | `validate google` | Validate Google OAuth2 credentials | | `reauth google` | Re-authenticate Google OAuth2 | -| `get-secret [--json] [--list]` | Read secret from provider store | +| `get-secret [--out FILE] [--json] [--list]` | Secret access (masked summary; --out writes to file) | | `list-providers` | List available API providers | | `init` | Initialize .env template at ~/.secrets/aipass/ | | `test` | Test OpenRouter connection status | @@ -49,8 +49,9 @@ drone @api [args] api/ ├── apps/ │ ├── api.py # Entry point — module discovery, command routing -│ ├── modules/ # Orchestration layer (7 modules) +│ ├── modules/ # Orchestration layer (8 modules) │ │ ├── api_key.py # Key retrieval, validation, provider listing +│ │ ├── secrets.py # Cross-branch secrets door (in-process API) │ │ ├── openrouter_client.py # OpenRouter client — calls, models, status │ │ ├── google_client.py # Google API services (Drive, Calendar, etc.) │ │ ├── usage_tracker.py # Usage metrics — track, stats, cleanup @@ -67,7 +68,7 @@ api/ │ │ └── usage/aggregation.py, cleanup.py, tracking.py │ └── integrations/ # Private driver space (gitignored) │ └── {project}/driver.py -└── tests/ # 499 tests across 28 files +└── tests/ # 504 tests across 28 files ``` Three-tier: entry point routes to modules (orchestration), modules delegate to handlers (business logic). Modules auto-discovered from `apps/modules/*.py` via `handle_command()`. @@ -87,10 +88,11 @@ service = get_drive_service(thread_safe=True) # For concurrent workers from aipass.api.apps.modules.google_client import get_google_service service = get_google_service("calendar", "v3") -from aipass.api.apps.handlers.auth.secrets import get_secret, list_secrets +from aipass.api.apps.modules.secrets import get_secret, list_secrets token = get_secret("telegram", "bot") # Returns bot_token string config = get_secret("telegram", "bot", as_json=True) # Returns full dict slugs = list_secrets("telegram") # Returns ["bot", "webhook", ...] +# CLI never prints raw values — use the Python API above for programmatic access ``` --- diff --git a/src/aipass/api/apps/modules/api_key.py b/src/aipass/api/apps/modules/api_key.py index b896b04b..2485d4a2 100644 --- a/src/aipass/api/apps/modules/api_key.py +++ b/src/aipass/api/apps/modules/api_key.py @@ -172,19 +172,32 @@ def init_env(): def get_secret_cmd(args: List[str]): - """Orchestrate secret retrieval workflow (machine-readable output)""" + """Orchestrate secret retrieval workflow (masked output only — no raw values to stdout)""" import json + import os if not args: - error("Usage: drone @api get-secret [--json] [--list]") + error("Usage: drone @api get-secret [--out FILE] [--json] [--list]") return has_json = "--json" in args has_list = "--list" in args + has_out = "--out" in args + out_file = None + if has_out: + out_idx = args.index("--out") + if out_idx + 1 < len(args): + out_file = args[out_idx + 1] + else: + error("--out requires a file path argument") + return + clean_args = [a for a in args if not a.startswith("--")] + if has_out and out_file in clean_args: + clean_args.remove(out_file) if not clean_args: - error("Usage: drone @api get-secret [--json] [--list]") + error("Usage: drone @api get-secret [--out FILE] [--json] [--list]") return parts = clean_args[0].split("/", 1) @@ -193,7 +206,8 @@ def get_secret_cmd(args: List[str]): if has_list: slugs = secrets.list_secrets(provider) for slug in slugs: - print(slug) + # codeql[py/clear-text-logging-sensitive-data] # slug names are identifiers, not secret values + console.print(slug) return if len(parts) != 2 or not parts[1]: @@ -201,19 +215,23 @@ def get_secret_cmd(args: List[str]): return slug = parts[1] + result = secrets.get_secret(provider, slug, as_json=has_json) - if has_json: - result = secrets.get_secret(provider, slug, as_json=True) - if result is not None: - print(json.dumps(result, indent=2)) - else: - error(f"Secret not found: {provider}/{slug}") + if result is None: + error(f"Secret not found: {provider}/{slug}") + return + + if out_file: + content = json.dumps(result, indent=2) if has_json else str(result) + fd = os.open(out_file, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600) + try: + os.write(fd, content.encode("utf-8")) + finally: + os.close(fd) + success(f"Wrote {provider}/{slug} to {out_file}") else: - result = secrets.get_secret(provider, slug) - if result is not None: - print(result) - else: - error(f"Secret not found: {provider}/{slug}") + value_len = len(json.dumps(result)) if has_json else len(str(result)) + success(f"{provider}/{slug}: set ({value_len} chars)") def fetch_api_key(provider: str = "openrouter"): @@ -255,12 +273,21 @@ EXAMPLES: # Get key for provider drone @api get-key openrouter - # Get secret for provider/slug + # Check if a secret exists (masked summary, no raw value) drone @api get-secret telegram/bot + # Write secret to a protected file + drone @api get-secret telegram/bot --out /tmp/token.txt + + # Write secret as JSON to a protected file + drone @api get-secret telegram/bot --out /tmp/bot.json --json + # List secrets for a provider drone @api get-secret telegram --list + # Programmatic access (in-process, no stdout): + # from aipass.api.apps.modules.secrets import get_secret + # Validate key drone @api validate openrouter diff --git a/src/aipass/api/apps/modules/secrets.py b/src/aipass/api/apps/modules/secrets.py new file mode 100644 index 00000000..327398a9 --- /dev/null +++ b/src/aipass/api/apps/modules/secrets.py @@ -0,0 +1,127 @@ +# =================== AIPass ==================== +# Name: secrets.py +# Description: Secrets Module — cross-branch in-process door +# Version: 1.0.0 +# Created: 2026-06-15 +# Modified: 2026-06-15 +# ============================================= + +""" +Secrets Module + +Cross-branch in-process API for reading secrets from the provider store. +Consumers import directly instead of shelling out to the CLI. + +Functions: + get_secret() - Read a secret by provider/slug + list_secrets() - List available slugs for a provider + handle_command() - Route CLI commands (seedgo module discovery) +""" + +import sys +from typing import Any, List, Optional + +from aipass.prax import logger # noqa: F401 — seedgo imports standard +from aipass.cli.apps.modules import console, header +from aipass.api.apps.handlers.json import json_handler +from aipass.api.apps.handlers.auth import secrets as _handler + + +def print_introspection(): + """Show module introspection - connected handlers and capabilities""" + console.print() + header("Secrets Module Introspection") + console.print() + + console.print("[cyan]Purpose:[/cyan] Cross-branch secrets access (in-process)") + console.print() + + console.print("[cyan]Connected Handlers:[/cyan]") + console.print(" • api.apps.handlers.auth.secrets") + console.print() + + console.print("[cyan]Available Workflows:[/cyan]") + console.print(" • get_secret() - Read secret by provider/slug") + console.print(" • list_secrets() - List slugs for a provider") + console.print() + + +def print_help(): + """Print help output for secrets module""" + print_introspection() + + +def handle_command(command: str, args: List[str]) -> bool: + """ + Handle secrets commands (module discovery hook). + + This module does not own any CLI commands — get-secret is routed + through api_key.py. This exists for seedgo module discovery only. + + Args: + command: Command name + args: Command arguments + + Returns: + False — no commands handled here + """ + if not args: + print_introspection() + return True + + if args[0] in ("--help", "-h", "help"): + print_help() + return True + + return False + + +def get_secret(provider: str, slug: str, as_json: bool = False) -> Optional[Any]: + """ + Read a secret from the provider store. + + This is the sanctioned cross-branch import path. Consumers call this + instead of shelling out to 'drone @api get-secret'. + + Args: + provider: Provider directory name (e.g., 'telegram', 'openrouter') + slug: Secret identifier (without .json extension) + as_json: If True, return full parsed dict; otherwise extract primary token + + Returns: + Secret value (str or dict) or None if not found + """ + result = _handler.get_secret(provider, slug, as_json=as_json) + json_handler.log_operation("secrets_get", {"provider": provider, "slug": slug, "found": result is not None}) + return result + + +def list_secrets(provider: str) -> List[str]: + """ + List available secret slugs for a provider. + + Args: + provider: Provider directory name + + Returns: + Sorted list of slug names + """ + return _handler.list_secrets(provider) + + +if __name__ == "__main__": + """Standalone execution mode""" + args = sys.argv[1:] + + if len(args) == 0: + print_introspection() + sys.exit(0) + + if args[0] in ["--help", "-h", "help"]: + print_help() + sys.exit(0) + + console.print() + console.print(f"[red]Unknown command: {args[0]}[/red]") + console.print() + sys.exit(1) diff --git a/src/aipass/api/tests/test_secrets.py b/src/aipass/api/tests/test_secrets.py index 3c91b38b..a4804592 100644 --- a/src/aipass/api/tests/test_secrets.py +++ b/src/aipass/api/tests/test_secrets.py @@ -6,9 +6,9 @@ # Modified: 2026-06-15 # ============================================= -"""Tests for apps/handlers/auth/secrets.py and apps/modules/api_key.get_secret_cmd. +"""Tests for apps/handlers/auth/secrets.py, apps/modules/secrets.py, and api_key.get_secret_cmd. -Tests — secrets.py (get_secret, list_secrets): +Tests — handlers/auth/secrets.py (get_secret, list_secrets): - get_secret: JSON token extraction via _TOKEN_KEYS - get_secret: as_json returns full parsed dict - get_secret: raw file fallback returns stripped content @@ -21,18 +21,27 @@ Tests — secrets.py (get_secret, list_secrets): - list_secrets: non-existent provider returns empty list - list_secrets: skips dotfiles, __pycache__, directories -Tests — api_key.py (get_secret_cmd): -- get_secret_cmd with provider/slug prints token -- get_secret_cmd with --json prints JSON -- get_secret_cmd with --list prints slugs -- get_secret_cmd with no args calls error() -- get_secret_cmd with provider only (no --list) calls error() -- get_secret_cmd with only flags (no positional args) calls error() +Tests — modules/secrets.py (in-process door): +- get_secret wraps handler and logs operation +- list_secrets wraps handler + +Tests — api_key.py (get_secret_cmd — hardened, no raw values to stdout): +- get_secret_cmd default prints masked summary only +- get_secret_cmd --out writes to file with 0o600 perms +- get_secret_cmd --out --json writes JSON to file +- get_secret_cmd --list prints slug names +- get_secret_cmd no args calls error() +- get_secret_cmd provider only (no --list) calls error() +- get_secret_cmd only flags calls error() +- get_secret_cmd not found calls error() +- get_secret_cmd --out missing path calls error() """ from __future__ import annotations import json +import os +import stat import sys from pathlib import Path from unittest.mock import patch, MagicMock @@ -40,11 +49,13 @@ from unittest.mock import patch, MagicMock import pytest from aipass.api.apps.modules.api_key import handle_command as _hc # noqa: F401 — seedgo test_coverage detection +from aipass.api.apps.modules.secrets import handle_command as _hc2 # noqa: F401 — seedgo test_coverage detection from aipass.api.apps.handlers.auth.secrets import ( get_secret, list_secrets, ) from aipass.api.apps.modules.api_key import get_secret_cmd +from aipass.api.apps.modules import secrets as secrets_module # Patch targets @@ -54,8 +65,13 @@ PATCH_LOGGER = "aipass.api.apps.handlers.auth.secrets.logger" PATCH_CMD_SECRETS = "aipass.api.apps.modules.api_key.secrets" PATCH_CMD_ERROR = "aipass.api.apps.modules.api_key.error" +PATCH_CMD_SUCCESS = "aipass.api.apps.modules.api_key.success" +PATCH_CMD_CONSOLE = "aipass.api.apps.modules.api_key.console" PATCH_CMD_JSON_HANDLER = "aipass.api.apps.modules.api_key.json_handler" +PATCH_MOD_HANDLER = "aipass.api.apps.modules.secrets._handler" +PATCH_MOD_JSON_HANDLER = "aipass.api.apps.modules.secrets.json_handler" + # ============================================= # get_secret @@ -291,47 +307,132 @@ class TestListSecrets: # ============================================= +class TestSecretsModule: + """Verifies the in-process module door (apps/modules/secrets.py).""" + + def test_get_secret_wraps_handler(self) -> None: + """Module get_secret delegates to handler and logs the operation.""" + mock_handler = MagicMock() + mock_handler.get_secret.return_value = "token123" + mock_jh = MagicMock() + + with patch(PATCH_MOD_HANDLER, mock_handler), patch(PATCH_MOD_JSON_HANDLER, mock_jh): + result = secrets_module.get_secret("telegram", "bot") + + assert result == "token123" + mock_handler.get_secret.assert_called_once_with("telegram", "bot", as_json=False) + mock_jh.log_operation.assert_called_once() + + def test_get_secret_as_json(self) -> None: + """Module get_secret passes as_json through to handler.""" + mock_handler = MagicMock() + data = {"bot_token": "abc"} + mock_handler.get_secret.return_value = data + mock_jh = MagicMock() + + with patch(PATCH_MOD_HANDLER, mock_handler), patch(PATCH_MOD_JSON_HANDLER, mock_jh): + result = secrets_module.get_secret("telegram", "bot", as_json=True) + + assert result == data + mock_handler.get_secret.assert_called_once_with("telegram", "bot", as_json=True) + + def test_get_secret_not_found_logs(self) -> None: + """Module get_secret logs even when handler returns None.""" + mock_handler = MagicMock() + mock_handler.get_secret.return_value = None + mock_jh = MagicMock() + + with patch(PATCH_MOD_HANDLER, mock_handler), patch(PATCH_MOD_JSON_HANDLER, mock_jh): + result = secrets_module.get_secret("telegram", "missing") + + assert result is None + log_call = mock_jh.log_operation.call_args + assert log_call[0][1]["found"] is False + + def test_list_secrets_wraps_handler(self) -> None: + """Module list_secrets delegates to handler.""" + mock_handler = MagicMock() + mock_handler.list_secrets.return_value = ["bot", "webhook"] + + with patch(PATCH_MOD_HANDLER, mock_handler): + result = secrets_module.list_secrets("telegram") + + assert result == ["bot", "webhook"] + mock_handler.list_secrets.assert_called_once_with("telegram") + + +# ============================================= +# get_secret_cmd (hardened — no raw values to stdout) +# ============================================= + + class TestGetSecretCmd: - """Verifies the get_secret_cmd orchestrator in api_key.py.""" + """Verifies the hardened get_secret_cmd (DPLAN-0211: no raw secrets to stdout).""" @patch(PATCH_CMD_JSON_HANDLER) @patch(PATCH_CMD_SECRETS) - @patch("builtins.print") - def test_prints_token(self, mock_print, mock_secrets, mock_jh) -> None: - """provider/slug prints the token to stdout.""" - mock_secrets.get_secret.return_value = "my-secret-token" + @patch(PATCH_CMD_SUCCESS) + def test_default_prints_masked_summary(self, mock_success, mock_secrets, mock_jh) -> None: + """Default (no flags) prints masked summary, never the raw value.""" + mock_secrets.get_secret.return_value = "my-secret-token-value" get_secret_cmd(["telegram/bot"]) - mock_secrets.get_secret.assert_called_once_with("telegram", "bot") - mock_print.assert_called_once_with("my-secret-token") + mock_secrets.get_secret.assert_called_once_with("telegram", "bot", as_json=False) + msg = mock_success.call_args[0][0] + assert "telegram/bot" in msg + assert "set" in msg + assert "chars" in msg + assert "my-secret-token-value" not in msg + @pytest.mark.skipif(sys.platform == "win32", reason="File permission checks are POSIX-only") @patch(PATCH_CMD_JSON_HANDLER) @patch(PATCH_CMD_SECRETS) - @patch("builtins.print") - def test_json_flag_prints_json(self, mock_print, mock_secrets, mock_jh) -> None: - """--json flag prints formatted JSON to stdout.""" - data = {"bot_token": "abc123", "webhook": "https://example.com"} + @patch(PATCH_CMD_SUCCESS) + def test_out_writes_file_with_0600(self, mock_success, mock_secrets, mock_jh, tmp_path: Path) -> None: + """--out writes secret value to file with 0o600 permissions.""" + mock_secrets.get_secret.return_value = "secret-token-here" + out_file = str(tmp_path / "token.txt") + + get_secret_cmd(["telegram/bot", "--out", out_file]) + + assert Path(out_file).exists() + assert Path(out_file).read_text(encoding="utf-8") == "secret-token-here" + file_mode = stat.S_IMODE(os.stat(out_file).st_mode) + assert file_mode == 0o600 + msg = mock_success.call_args[0][0] + assert out_file in msg + assert "secret-token-here" not in msg + + @pytest.mark.skipif(sys.platform == "win32", reason="File permission checks are POSIX-only") + @patch(PATCH_CMD_JSON_HANDLER) + @patch(PATCH_CMD_SECRETS) + @patch(PATCH_CMD_SUCCESS) + def test_out_json_writes_json_file(self, mock_success, mock_secrets, mock_jh, tmp_path: Path) -> None: + """--out --json writes JSON-formatted secret to file.""" + data = {"bot_token": "abc123", "allowed": [1, 2]} mock_secrets.get_secret.return_value = data + out_file = str(tmp_path / "bot.json") - get_secret_cmd(["telegram/bot", "--json"]) + get_secret_cmd(["telegram/bot", "--out", out_file, "--json"]) - mock_secrets.get_secret.assert_called_once_with("telegram", "bot", as_json=True) - mock_print.assert_called_once_with(json.dumps(data, indent=2)) + content = Path(out_file).read_text(encoding="utf-8") + assert json.loads(content) == data + file_mode = stat.S_IMODE(os.stat(out_file).st_mode) + assert file_mode == 0o600 @patch(PATCH_CMD_JSON_HANDLER) @patch(PATCH_CMD_SECRETS) - @patch("builtins.print") - def test_list_flag_prints_slugs(self, mock_print, mock_secrets, mock_jh) -> None: - """--list flag prints each slug on its own line.""" + @patch(PATCH_CMD_CONSOLE) + def test_list_prints_slugs(self, mock_console, mock_secrets, mock_jh) -> None: + """--list prints slug names via console.print.""" mock_secrets.list_secrets.return_value = ["bot", "webhook"] get_secret_cmd(["telegram", "--list"]) mock_secrets.list_secrets.assert_called_once_with("telegram") - assert mock_print.call_count == 2 - mock_print.assert_any_call("bot") - mock_print.assert_any_call("webhook") + calls = [c for c in mock_console.print.call_args_list if c[0][0] in ("bot", "webhook")] + assert len(calls) == 2 @patch(PATCH_CMD_JSON_HANDLER) @patch(PATCH_CMD_ERROR) @@ -349,7 +450,6 @@ class TestGetSecretCmd: get_secret_cmd(["telegram"]) mock_error.assert_called_once() - assert "provider" in mock_error.call_args[0][0].lower() or "format" in mock_error.call_args[0][0].lower() @patch(PATCH_CMD_JSON_HANDLER) @patch(PATCH_CMD_ERROR) @@ -373,25 +473,21 @@ class TestGetSecretCmd: assert "not found" in mock_error.call_args[0][0].lower() @patch(PATCH_CMD_JSON_HANDLER) - @patch(PATCH_CMD_SECRETS) @patch(PATCH_CMD_ERROR) - def test_json_secret_not_found_calls_error(self, mock_error, mock_secrets, mock_jh) -> None: - """When get_secret with --json returns None, error() is called.""" - mock_secrets.get_secret.return_value = None - - get_secret_cmd(["telegram/bot", "--json"]) + def test_out_missing_path_calls_error(self, mock_error, mock_jh) -> None: + """--out without a file path argument calls error().""" + get_secret_cmd(["telegram/bot", "--out"]) mock_error.assert_called_once() - assert "not found" in mock_error.call_args[0][0].lower() + assert "--out" in mock_error.call_args[0][0] @patch(PATCH_CMD_JSON_HANDLER) @patch(PATCH_CMD_SECRETS) - @patch("builtins.print") - def test_list_empty_provider(self, mock_print, mock_secrets, mock_jh) -> None: + @patch(PATCH_CMD_CONSOLE) + def test_list_empty_provider(self, mock_console, mock_secrets, mock_jh) -> None: """--list with provider that has no secrets prints nothing.""" mock_secrets.list_secrets.return_value = [] get_secret_cmd(["empty_provider", "--list"]) mock_secrets.list_secrets.assert_called_once_with("empty_provider") - mock_print.assert_not_called() diff --git a/src/aipass/seedgo/tests/test_checkers_batch2.py b/src/aipass/seedgo/tests/test_checkers_batch2.py index da84229d..c9c98e12 100644 --- a/src/aipass/seedgo/tests/test_checkers_batch2.py +++ b/src/aipass/seedgo/tests/test_checkers_batch2.py @@ -181,6 +181,8 @@ def call_api(): assert result["standard"] == "HARDCODED_KEY" def test_hardcoded_key_violation_caught(self, tmp_path: Path) -> None: + # NOTE: the sk-or-v1-... literal below is a FAKE/synthetic key (patterned hex, + # not a real credential). It exists only to prove the detector flags hardcoded keys. code = """\ API_KEY = "sk-or-v1-9f8e7d6c5b4a3f2e1d0c9b8a7f6e5d4c3b2a1f0e" @@ -195,6 +197,7 @@ def call_api(): assert "hardcoded" in violations[0]["message"].lower() or "key" in violations[0]["message"].lower() def test_hardcoded_key_bypass_respected(self, tmp_path: Path) -> None: + # NOTE: same FAKE/synthetic sk-or-v1-... fixture key below — not a real credential. code = """\ API_KEY = "sk-or-v1-9f8e7d6c5b4a3f2e1d0c9b8a7f6e5d4c3b2a1f0e" """ diff --git a/src/aipass/skills/.aipass/skills/telegram/SKILL.md b/src/aipass/skills/.aipass/skills/telegram/SKILL.md index acdd8dbe..013ee434 100644 --- a/src/aipass/skills/.aipass/skills/telegram/SKILL.md +++ b/src/aipass/skills/.aipass/skills/telegram/SKILL.md @@ -44,5 +44,5 @@ drone @skills run telegram notify "message" ## Secrets -Bot tokens and config accessed via `drone @api get-secret telegram/`. +Bot tokens and config accessed via the in-process `aipass.api.apps.modules.secrets.get_secret` API. State files (offset, lock, registry) stay with the skill in `.local/`. diff --git a/src/aipass/skills/.aipass/skills/telegram/apps/handlers/config.py b/src/aipass/skills/.aipass/skills/telegram/apps/handlers/config.py index a32ded90..af21e76d 100644 --- a/src/aipass/skills/.aipass/skills/telegram/apps/handlers/config.py +++ b/src/aipass/skills/.aipass/skills/telegram/apps/handlers/config.py @@ -1,12 +1,14 @@ # Standard library -import json -import subprocess from pathlib import Path from typing import Optional, List # Logging from aipass.prax import logger +# Cross-branch in-process secrets API +from aipass.api.apps.modules.secrets import get_secret as _api_get_secret +from aipass.api.apps.modules.secrets import list_secrets as _api_list_secrets + # ============================================= # CONSTANTS # ============================================= @@ -14,7 +16,7 @@ from aipass.prax import logger REQUIRED_BOT_FIELDS = ("bot_id", "bot_token") # ============================================= -# SECRETS ACCESS (via drone @api) +# SECRETS ACCESS (in-process @api) # ============================================= @@ -22,8 +24,8 @@ def _get_secret(bot_id: str) -> dict | None: """ Retrieve bot config from the API secrets store. - Calls `drone @api get-secret telegram/ --json` via subprocess - and returns the parsed JSON config dict. + Uses the in-process aipass.api.apps.modules.secrets.get_secret API + (no subprocess, no stdout parsing, no token leakage). Args: bot_id: Bot identifier to look up. @@ -32,34 +34,16 @@ def _get_secret(bot_id: str) -> dict | None: Config dict or None if the call fails or returns no data. """ try: - result = subprocess.run( - ["drone", "@api", "get-secret", f"telegram/{bot_id}", "--json"], - capture_output=True, - text=True, - timeout=10, - ) - if result.returncode != 0: - logger.warning(f"drone @api get-secret telegram/{bot_id} failed: {result.stderr.strip()}") + result = _api_get_secret("telegram", bot_id, as_json=True) + if result is None: + logger.warning("Secret not found: telegram/%s", bot_id) return None - - raw = result.stdout.strip() - if not raw: + if not isinstance(result, dict): + logger.warning("Secret telegram/%s is not a dict", bot_id) return None - - config = json.loads(raw) - if not isinstance(config, dict): - return None - - return config - - except subprocess.TimeoutExpired: - logger.error(f"Timeout fetching secret for bot_id={bot_id}") - return None - except json.JSONDecodeError as e: - logger.error(f"Invalid JSON from get-secret telegram/{bot_id}: {e}") - return None + return result except Exception as e: - logger.error(f"Unexpected error fetching secret for bot_id={bot_id}: {e}") + logger.error("Failed to fetch secret telegram/%s: %s", bot_id, e) return None @@ -160,7 +144,7 @@ def load_bot_config(bot_id: str) -> dict | None: """ Load per-bot config from the API secrets store. - Fetches `drone @api get-secret telegram/ --json`. + Uses the in-process secrets API: get_secret("telegram", bot_id). Config format: { @@ -185,41 +169,15 @@ def list_bot_configs() -> list[str]: """ List all registered bot IDs via the API secrets store. - Calls `drone @api get-secret telegram --list` and parses the output - as a JSON list of bot_id strings. + Uses the in-process aipass.api.apps.modules.secrets.list_secrets API. Returns: List of bot_id strings, or empty list on failure. """ try: - result = subprocess.run( - ["drone", "@api", "get-secret", "telegram", "--list"], - capture_output=True, - text=True, - timeout=10, - ) - if result.returncode != 0: - logger.warning(f"drone @api get-secret telegram --list failed: {result.stderr.strip()}") - return [] - - raw = result.stdout.strip() - if not raw: - return [] - - bot_ids = json.loads(raw) - if not isinstance(bot_ids, list): - return [] - - return [str(b) for b in bot_ids] - - except subprocess.TimeoutExpired: - logger.error("Timeout listing telegram bot secrets") - return [] - except json.JSONDecodeError as e: - logger.error(f"Invalid JSON from get-secret telegram --list: {e}") - return [] + return _api_list_secrets("telegram") except Exception as e: - logger.error(f"Unexpected error listing telegram bot secrets: {e}") + logger.error("Failed to list telegram secrets: %s", e) return [] diff --git a/src/aipass/skills/.aipass/skills/telegram/tests/test_multibot_config.py b/src/aipass/skills/.aipass/skills/telegram/tests/test_multibot_config.py index 09c7c5d5..a2197524 100644 --- a/src/aipass/skills/.aipass/skills/telegram/tests/test_multibot_config.py +++ b/src/aipass/skills/.aipass/skills/telegram/tests/test_multibot_config.py @@ -7,7 +7,6 @@ Covers: - Bot operations (bot_operations.py): parse_create_args, format_bot_details, format_bot_table """ -import json from unittest.mock import patch, MagicMock import pytest @@ -122,17 +121,12 @@ class TestLoadBotConfig: class TestListBotConfigs: - """Tests for config.list_bot_configs (via subprocess).""" + """Tests for config.list_bot_configs (via in-process secrets API).""" - @patch("apps.handlers.config.subprocess") - def test_list_returns_bot_ids(self, mock_subprocess: MagicMock) -> None: - """Returns list of bot_ids from subprocess output.""" - mock_result = MagicMock() - mock_result.returncode = 0 - mock_result.stdout = json.dumps(["dev_central", "assistant", "scheduler"]) - mock_result.stderr = "" - mock_subprocess.run.return_value = mock_result - mock_subprocess.TimeoutExpired = TimeoutError + @patch("apps.handlers.config._api_list_secrets") + def test_list_returns_bot_ids(self, mock_list: MagicMock) -> None: + """Returns list of bot_ids from the secrets API.""" + mock_list.return_value = ["dev_central", "assistant", "scheduler"] result = tg_config.list_bot_configs() assert isinstance(result, list) @@ -140,42 +134,28 @@ class TestListBotConfigs: assert "assistant" in result assert "scheduler" in result assert len(result) == 3 + mock_list.assert_called_once_with("telegram") - @patch("apps.handlers.config.subprocess") - def test_list_returns_empty_on_failure(self, mock_subprocess: MagicMock) -> None: - """Returns empty list when subprocess fails.""" - mock_result = MagicMock() - mock_result.returncode = 1 - mock_result.stdout = "" - mock_result.stderr = "error" - mock_subprocess.run.return_value = mock_result - mock_subprocess.TimeoutExpired = TimeoutError + @patch("apps.handlers.config._api_list_secrets") + def test_list_returns_empty_on_failure(self, mock_list: MagicMock) -> None: + """Returns empty list when the secrets API raises.""" + mock_list.side_effect = RuntimeError("connection failed") result = tg_config.list_bot_configs() assert result == [] - @patch("apps.handlers.config.subprocess") - def test_list_returns_empty_on_empty_output(self, mock_subprocess: MagicMock) -> None: - """Returns empty list when subprocess returns empty output.""" - mock_result = MagicMock() - mock_result.returncode = 0 - mock_result.stdout = "" - mock_result.stderr = "" - mock_subprocess.run.return_value = mock_result - mock_subprocess.TimeoutExpired = TimeoutError + @patch("apps.handlers.config._api_list_secrets") + def test_list_returns_empty_when_no_secrets(self, mock_list: MagicMock) -> None: + """Returns empty list when no secrets exist.""" + mock_list.return_value = [] result = tg_config.list_bot_configs() assert result == [] - @patch("apps.handlers.config.subprocess") - def test_list_returns_empty_on_invalid_json(self, mock_subprocess: MagicMock) -> None: - """Returns empty list when subprocess returns invalid JSON.""" - mock_result = MagicMock() - mock_result.returncode = 0 - mock_result.stdout = "not valid json" - mock_result.stderr = "" - mock_subprocess.run.return_value = mock_result - mock_subprocess.TimeoutExpired = TimeoutError + @patch("apps.handlers.config._api_list_secrets") + def test_list_returns_empty_on_unexpected_error(self, mock_list: MagicMock) -> None: + """Returns empty list on unexpected exception.""" + mock_list.side_effect = OSError("disk error") result = tg_config.list_bot_configs() assert result == []