fix(api,skills): harden secrets door — no secret value to stdout (DPLAN-0211, clears CodeQL #86-88)
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) <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Opus 4.8
parent
c8ae084f54
commit
a75910a4e6
@@ -133,13 +133,7 @@
|
|||||||
{
|
{
|
||||||
"file": "apps/modules/api_key.py",
|
"file": "apps/modules/api_key.py",
|
||||||
"standard": "cli",
|
"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."
|
"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": "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."
|
|
||||||
},
|
},
|
||||||
{
|
{
|
||||||
"file": "tests/test_aggregation.py",
|
"file": "tests/test_aggregation.py",
|
||||||
|
|||||||
@@ -5,7 +5,7 @@
|
|||||||
> Centralized external API gateway — authenticated service clients for all external APIs
|
> Centralized external API gateway — authenticated service clients for all external APIs
|
||||||
|
|
||||||
**Module:** `aipass.api` | **Role:** `api_gateway`
|
**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
|
**Last Updated:** 2026-06-15
|
||||||
|
|
||||||
---
|
---
|
||||||
@@ -26,7 +26,7 @@ drone @api <command> [args]
|
|||||||
| `validate [provider]` | Validate API key (default: openrouter) |
|
| `validate [provider]` | Validate API key (default: openrouter) |
|
||||||
| `validate google` | Validate Google OAuth2 credentials |
|
| `validate google` | Validate Google OAuth2 credentials |
|
||||||
| `reauth google` | Re-authenticate Google OAuth2 |
|
| `reauth google` | Re-authenticate Google OAuth2 |
|
||||||
| `get-secret <provider/slug> [--json] [--list]` | Read secret from provider store |
|
| `get-secret <provider/slug> [--out FILE] [--json] [--list]` | Secret access (masked summary; --out writes to file) |
|
||||||
| `list-providers` | List available API providers |
|
| `list-providers` | List available API providers |
|
||||||
| `init` | Initialize .env template at ~/.secrets/aipass/ |
|
| `init` | Initialize .env template at ~/.secrets/aipass/ |
|
||||||
| `test` | Test OpenRouter connection status |
|
| `test` | Test OpenRouter connection status |
|
||||||
@@ -49,8 +49,9 @@ drone @api <command> [args]
|
|||||||
api/
|
api/
|
||||||
├── apps/
|
├── apps/
|
||||||
│ ├── api.py # Entry point — module discovery, command routing
|
│ ├── 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
|
│ │ ├── 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
|
│ │ ├── openrouter_client.py # OpenRouter client — calls, models, status
|
||||||
│ │ ├── google_client.py # Google API services (Drive, Calendar, etc.)
|
│ │ ├── google_client.py # Google API services (Drive, Calendar, etc.)
|
||||||
│ │ ├── usage_tracker.py # Usage metrics — track, stats, cleanup
|
│ │ ├── usage_tracker.py # Usage metrics — track, stats, cleanup
|
||||||
@@ -67,7 +68,7 @@ api/
|
|||||||
│ │ └── usage/aggregation.py, cleanup.py, tracking.py
|
│ │ └── usage/aggregation.py, cleanup.py, tracking.py
|
||||||
│ └── integrations/ # Private driver space (gitignored)
|
│ └── integrations/ # Private driver space (gitignored)
|
||||||
│ └── {project}/driver.py
|
│ └── {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()`.
|
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
|
from aipass.api.apps.modules.google_client import get_google_service
|
||||||
service = get_google_service("calendar", "v3")
|
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
|
token = get_secret("telegram", "bot") # Returns bot_token string
|
||||||
config = get_secret("telegram", "bot", as_json=True) # Returns full dict
|
config = get_secret("telegram", "bot", as_json=True) # Returns full dict
|
||||||
slugs = list_secrets("telegram") # Returns ["bot", "webhook", ...]
|
slugs = list_secrets("telegram") # Returns ["bot", "webhook", ...]
|
||||||
|
# CLI never prints raw values — use the Python API above for programmatic access
|
||||||
```
|
```
|
||||||
|
|
||||||
---
|
---
|
||||||
|
|||||||
@@ -172,19 +172,32 @@ def init_env():
|
|||||||
|
|
||||||
|
|
||||||
def get_secret_cmd(args: List[str]):
|
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 json
|
||||||
|
import os
|
||||||
|
|
||||||
if not args:
|
if not args:
|
||||||
error("Usage: drone @api get-secret <provider/slug> [--json] [--list]")
|
error("Usage: drone @api get-secret <provider/slug> [--out FILE] [--json] [--list]")
|
||||||
return
|
return
|
||||||
|
|
||||||
has_json = "--json" in args
|
has_json = "--json" in args
|
||||||
has_list = "--list" 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("--")]
|
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:
|
if not clean_args:
|
||||||
error("Usage: drone @api get-secret <provider/slug> [--json] [--list]")
|
error("Usage: drone @api get-secret <provider/slug> [--out FILE] [--json] [--list]")
|
||||||
return
|
return
|
||||||
|
|
||||||
parts = clean_args[0].split("/", 1)
|
parts = clean_args[0].split("/", 1)
|
||||||
@@ -193,7 +206,8 @@ def get_secret_cmd(args: List[str]):
|
|||||||
if has_list:
|
if has_list:
|
||||||
slugs = secrets.list_secrets(provider)
|
slugs = secrets.list_secrets(provider)
|
||||||
for slug in slugs:
|
for slug in slugs:
|
||||||
print(slug)
|
# codeql[py/clear-text-logging-sensitive-data] # slug names are identifiers, not secret values
|
||||||
|
console.print(slug)
|
||||||
return
|
return
|
||||||
|
|
||||||
if len(parts) != 2 or not parts[1]:
|
if len(parts) != 2 or not parts[1]:
|
||||||
@@ -201,19 +215,23 @@ def get_secret_cmd(args: List[str]):
|
|||||||
return
|
return
|
||||||
|
|
||||||
slug = parts[1]
|
slug = parts[1]
|
||||||
|
result = secrets.get_secret(provider, slug, as_json=has_json)
|
||||||
|
|
||||||
if has_json:
|
if result is None:
|
||||||
result = secrets.get_secret(provider, slug, as_json=True)
|
error(f"Secret not found: {provider}/{slug}")
|
||||||
if result is not None:
|
return
|
||||||
print(json.dumps(result, indent=2))
|
|
||||||
else:
|
if out_file:
|
||||||
error(f"Secret not found: {provider}/{slug}")
|
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:
|
else:
|
||||||
result = secrets.get_secret(provider, slug)
|
value_len = len(json.dumps(result)) if has_json else len(str(result))
|
||||||
if result is not None:
|
success(f"{provider}/{slug}: set ({value_len} chars)")
|
||||||
print(result)
|
|
||||||
else:
|
|
||||||
error(f"Secret not found: {provider}/{slug}")
|
|
||||||
|
|
||||||
|
|
||||||
def fetch_api_key(provider: str = "openrouter"):
|
def fetch_api_key(provider: str = "openrouter"):
|
||||||
@@ -255,12 +273,21 @@ EXAMPLES:
|
|||||||
# Get key for provider
|
# Get key for provider
|
||||||
drone @api get-key openrouter
|
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
|
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
|
# List secrets for a provider
|
||||||
drone @api get-secret telegram --list
|
drone @api get-secret telegram --list
|
||||||
|
|
||||||
|
# Programmatic access (in-process, no stdout):
|
||||||
|
# from aipass.api.apps.modules.secrets import get_secret
|
||||||
|
|
||||||
# Validate key
|
# Validate key
|
||||||
drone @api validate openrouter
|
drone @api validate openrouter
|
||||||
|
|
||||||
|
|||||||
@@ -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)
|
||||||
@@ -6,9 +6,9 @@
|
|||||||
# Modified: 2026-06-15
|
# 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: JSON token extraction via _TOKEN_KEYS
|
||||||
- get_secret: as_json returns full parsed dict
|
- get_secret: as_json returns full parsed dict
|
||||||
- get_secret: raw file fallback returns stripped content
|
- 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: non-existent provider returns empty list
|
||||||
- list_secrets: skips dotfiles, __pycache__, directories
|
- list_secrets: skips dotfiles, __pycache__, directories
|
||||||
|
|
||||||
Tests — api_key.py (get_secret_cmd):
|
Tests — modules/secrets.py (in-process door):
|
||||||
- get_secret_cmd with provider/slug prints token
|
- get_secret wraps handler and logs operation
|
||||||
- get_secret_cmd with --json prints JSON
|
- list_secrets wraps handler
|
||||||
- get_secret_cmd with --list prints slugs
|
|
||||||
- get_secret_cmd with no args calls error()
|
Tests — api_key.py (get_secret_cmd — hardened, no raw values to stdout):
|
||||||
- get_secret_cmd with provider only (no --list) calls error()
|
- get_secret_cmd default prints masked summary only
|
||||||
- get_secret_cmd with only flags (no positional args) calls error()
|
- 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
|
from __future__ import annotations
|
||||||
|
|
||||||
import json
|
import json
|
||||||
|
import os
|
||||||
|
import stat
|
||||||
import sys
|
import sys
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from unittest.mock import patch, MagicMock
|
from unittest.mock import patch, MagicMock
|
||||||
@@ -40,11 +49,13 @@ from unittest.mock import patch, MagicMock
|
|||||||
import pytest
|
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.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 (
|
from aipass.api.apps.handlers.auth.secrets import (
|
||||||
get_secret,
|
get_secret,
|
||||||
list_secrets,
|
list_secrets,
|
||||||
)
|
)
|
||||||
from aipass.api.apps.modules.api_key import get_secret_cmd
|
from aipass.api.apps.modules.api_key import get_secret_cmd
|
||||||
|
from aipass.api.apps.modules import secrets as secrets_module
|
||||||
|
|
||||||
|
|
||||||
# Patch targets
|
# 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_SECRETS = "aipass.api.apps.modules.api_key.secrets"
|
||||||
PATCH_CMD_ERROR = "aipass.api.apps.modules.api_key.error"
|
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_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
|
# 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:
|
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_JSON_HANDLER)
|
||||||
@patch(PATCH_CMD_SECRETS)
|
@patch(PATCH_CMD_SECRETS)
|
||||||
@patch("builtins.print")
|
@patch(PATCH_CMD_SUCCESS)
|
||||||
def test_prints_token(self, mock_print, mock_secrets, mock_jh) -> None:
|
def test_default_prints_masked_summary(self, mock_success, mock_secrets, mock_jh) -> None:
|
||||||
"""provider/slug prints the token to stdout."""
|
"""Default (no flags) prints masked summary, never the raw value."""
|
||||||
mock_secrets.get_secret.return_value = "my-secret-token"
|
mock_secrets.get_secret.return_value = "my-secret-token-value"
|
||||||
|
|
||||||
get_secret_cmd(["telegram/bot"])
|
get_secret_cmd(["telegram/bot"])
|
||||||
|
|
||||||
mock_secrets.get_secret.assert_called_once_with("telegram", "bot")
|
mock_secrets.get_secret.assert_called_once_with("telegram", "bot", as_json=False)
|
||||||
mock_print.assert_called_once_with("my-secret-token")
|
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_JSON_HANDLER)
|
||||||
@patch(PATCH_CMD_SECRETS)
|
@patch(PATCH_CMD_SECRETS)
|
||||||
@patch("builtins.print")
|
@patch(PATCH_CMD_SUCCESS)
|
||||||
def test_json_flag_prints_json(self, mock_print, mock_secrets, mock_jh) -> None:
|
def test_out_writes_file_with_0600(self, mock_success, mock_secrets, mock_jh, tmp_path: Path) -> None:
|
||||||
"""--json flag prints formatted JSON to stdout."""
|
"""--out writes secret value to file with 0o600 permissions."""
|
||||||
data = {"bot_token": "abc123", "webhook": "https://example.com"}
|
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
|
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)
|
content = Path(out_file).read_text(encoding="utf-8")
|
||||||
mock_print.assert_called_once_with(json.dumps(data, indent=2))
|
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_JSON_HANDLER)
|
||||||
@patch(PATCH_CMD_SECRETS)
|
@patch(PATCH_CMD_SECRETS)
|
||||||
@patch("builtins.print")
|
@patch(PATCH_CMD_CONSOLE)
|
||||||
def test_list_flag_prints_slugs(self, mock_print, mock_secrets, mock_jh) -> None:
|
def test_list_prints_slugs(self, mock_console, mock_secrets, mock_jh) -> None:
|
||||||
"""--list flag prints each slug on its own line."""
|
"""--list prints slug names via console.print."""
|
||||||
mock_secrets.list_secrets.return_value = ["bot", "webhook"]
|
mock_secrets.list_secrets.return_value = ["bot", "webhook"]
|
||||||
|
|
||||||
get_secret_cmd(["telegram", "--list"])
|
get_secret_cmd(["telegram", "--list"])
|
||||||
|
|
||||||
mock_secrets.list_secrets.assert_called_once_with("telegram")
|
mock_secrets.list_secrets.assert_called_once_with("telegram")
|
||||||
assert mock_print.call_count == 2
|
calls = [c for c in mock_console.print.call_args_list if c[0][0] in ("bot", "webhook")]
|
||||||
mock_print.assert_any_call("bot")
|
assert len(calls) == 2
|
||||||
mock_print.assert_any_call("webhook")
|
|
||||||
|
|
||||||
@patch(PATCH_CMD_JSON_HANDLER)
|
@patch(PATCH_CMD_JSON_HANDLER)
|
||||||
@patch(PATCH_CMD_ERROR)
|
@patch(PATCH_CMD_ERROR)
|
||||||
@@ -349,7 +450,6 @@ class TestGetSecretCmd:
|
|||||||
get_secret_cmd(["telegram"])
|
get_secret_cmd(["telegram"])
|
||||||
|
|
||||||
mock_error.assert_called_once()
|
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_JSON_HANDLER)
|
||||||
@patch(PATCH_CMD_ERROR)
|
@patch(PATCH_CMD_ERROR)
|
||||||
@@ -373,25 +473,21 @@ class TestGetSecretCmd:
|
|||||||
assert "not found" in mock_error.call_args[0][0].lower()
|
assert "not found" in mock_error.call_args[0][0].lower()
|
||||||
|
|
||||||
@patch(PATCH_CMD_JSON_HANDLER)
|
@patch(PATCH_CMD_JSON_HANDLER)
|
||||||
@patch(PATCH_CMD_SECRETS)
|
|
||||||
@patch(PATCH_CMD_ERROR)
|
@patch(PATCH_CMD_ERROR)
|
||||||
def test_json_secret_not_found_calls_error(self, mock_error, mock_secrets, mock_jh) -> None:
|
def test_out_missing_path_calls_error(self, mock_error, mock_jh) -> None:
|
||||||
"""When get_secret with --json returns None, error() is called."""
|
"""--out without a file path argument calls error()."""
|
||||||
mock_secrets.get_secret.return_value = None
|
get_secret_cmd(["telegram/bot", "--out"])
|
||||||
|
|
||||||
get_secret_cmd(["telegram/bot", "--json"])
|
|
||||||
|
|
||||||
mock_error.assert_called_once()
|
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_JSON_HANDLER)
|
||||||
@patch(PATCH_CMD_SECRETS)
|
@patch(PATCH_CMD_SECRETS)
|
||||||
@patch("builtins.print")
|
@patch(PATCH_CMD_CONSOLE)
|
||||||
def test_list_empty_provider(self, mock_print, mock_secrets, mock_jh) -> None:
|
def test_list_empty_provider(self, mock_console, mock_secrets, mock_jh) -> None:
|
||||||
"""--list with provider that has no secrets prints nothing."""
|
"""--list with provider that has no secrets prints nothing."""
|
||||||
mock_secrets.list_secrets.return_value = []
|
mock_secrets.list_secrets.return_value = []
|
||||||
|
|
||||||
get_secret_cmd(["empty_provider", "--list"])
|
get_secret_cmd(["empty_provider", "--list"])
|
||||||
|
|
||||||
mock_secrets.list_secrets.assert_called_once_with("empty_provider")
|
mock_secrets.list_secrets.assert_called_once_with("empty_provider")
|
||||||
mock_print.assert_not_called()
|
|
||||||
|
|||||||
@@ -181,6 +181,8 @@ def call_api():
|
|||||||
assert result["standard"] == "HARDCODED_KEY"
|
assert result["standard"] == "HARDCODED_KEY"
|
||||||
|
|
||||||
def test_hardcoded_key_violation_caught(self, tmp_path: Path) -> None:
|
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 = """\
|
code = """\
|
||||||
API_KEY = "sk-or-v1-9f8e7d6c5b4a3f2e1d0c9b8a7f6e5d4c3b2a1f0e"
|
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()
|
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:
|
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 = """\
|
code = """\
|
||||||
API_KEY = "sk-or-v1-9f8e7d6c5b4a3f2e1d0c9b8a7f6e5d4c3b2a1f0e"
|
API_KEY = "sk-or-v1-9f8e7d6c5b4a3f2e1d0c9b8a7f6e5d4c3b2a1f0e"
|
||||||
"""
|
"""
|
||||||
|
|||||||
@@ -44,5 +44,5 @@ drone @skills run telegram notify "message"
|
|||||||
|
|
||||||
## Secrets
|
## Secrets
|
||||||
|
|
||||||
Bot tokens and config accessed via `drone @api get-secret telegram/<bot_id>`.
|
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/`.
|
State files (offset, lock, registry) stay with the skill in `.local/`.
|
||||||
|
|||||||
@@ -1,12 +1,14 @@
|
|||||||
# Standard library
|
# Standard library
|
||||||
import json
|
|
||||||
import subprocess
|
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from typing import Optional, List
|
from typing import Optional, List
|
||||||
|
|
||||||
# Logging
|
# Logging
|
||||||
from aipass.prax import logger
|
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
|
# CONSTANTS
|
||||||
# =============================================
|
# =============================================
|
||||||
@@ -14,7 +16,7 @@ from aipass.prax import logger
|
|||||||
REQUIRED_BOT_FIELDS = ("bot_id", "bot_token")
|
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.
|
Retrieve bot config from the API secrets store.
|
||||||
|
|
||||||
Calls `drone @api get-secret telegram/<bot_id> --json` via subprocess
|
Uses the in-process aipass.api.apps.modules.secrets.get_secret API
|
||||||
and returns the parsed JSON config dict.
|
(no subprocess, no stdout parsing, no token leakage).
|
||||||
|
|
||||||
Args:
|
Args:
|
||||||
bot_id: Bot identifier to look up.
|
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.
|
Config dict or None if the call fails or returns no data.
|
||||||
"""
|
"""
|
||||||
try:
|
try:
|
||||||
result = subprocess.run(
|
result = _api_get_secret("telegram", bot_id, as_json=True)
|
||||||
["drone", "@api", "get-secret", f"telegram/{bot_id}", "--json"],
|
if result is None:
|
||||||
capture_output=True,
|
logger.warning("Secret not found: telegram/%s", bot_id)
|
||||||
text=True,
|
|
||||||
timeout=10,
|
|
||||||
)
|
|
||||||
if result.returncode != 0:
|
|
||||||
logger.warning(f"drone @api get-secret telegram/{bot_id} failed: {result.stderr.strip()}")
|
|
||||||
return None
|
return None
|
||||||
|
if not isinstance(result, dict):
|
||||||
raw = result.stdout.strip()
|
logger.warning("Secret telegram/%s is not a dict", bot_id)
|
||||||
if not raw:
|
|
||||||
return None
|
return None
|
||||||
|
return result
|
||||||
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
|
|
||||||
except Exception as e:
|
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
|
return None
|
||||||
|
|
||||||
|
|
||||||
@@ -160,7 +144,7 @@ def load_bot_config(bot_id: str) -> dict | None:
|
|||||||
"""
|
"""
|
||||||
Load per-bot config from the API secrets store.
|
Load per-bot config from the API secrets store.
|
||||||
|
|
||||||
Fetches `drone @api get-secret telegram/<bot_id> --json`.
|
Uses the in-process secrets API: get_secret("telegram", bot_id).
|
||||||
|
|
||||||
Config format:
|
Config format:
|
||||||
{
|
{
|
||||||
@@ -185,41 +169,15 @@ def list_bot_configs() -> list[str]:
|
|||||||
"""
|
"""
|
||||||
List all registered bot IDs via the API secrets store.
|
List all registered bot IDs via the API secrets store.
|
||||||
|
|
||||||
Calls `drone @api get-secret telegram --list` and parses the output
|
Uses the in-process aipass.api.apps.modules.secrets.list_secrets API.
|
||||||
as a JSON list of bot_id strings.
|
|
||||||
|
|
||||||
Returns:
|
Returns:
|
||||||
List of bot_id strings, or empty list on failure.
|
List of bot_id strings, or empty list on failure.
|
||||||
"""
|
"""
|
||||||
try:
|
try:
|
||||||
result = subprocess.run(
|
return _api_list_secrets("telegram")
|
||||||
["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 []
|
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
logger.error(f"Unexpected error listing telegram bot secrets: {e}")
|
logger.error("Failed to list telegram secrets: %s", e)
|
||||||
return []
|
return []
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@@ -7,7 +7,6 @@ Covers:
|
|||||||
- Bot operations (bot_operations.py): parse_create_args, format_bot_details, format_bot_table
|
- Bot operations (bot_operations.py): parse_create_args, format_bot_details, format_bot_table
|
||||||
"""
|
"""
|
||||||
|
|
||||||
import json
|
|
||||||
from unittest.mock import patch, MagicMock
|
from unittest.mock import patch, MagicMock
|
||||||
|
|
||||||
import pytest
|
import pytest
|
||||||
@@ -122,17 +121,12 @@ class TestLoadBotConfig:
|
|||||||
|
|
||||||
|
|
||||||
class TestListBotConfigs:
|
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")
|
@patch("apps.handlers.config._api_list_secrets")
|
||||||
def test_list_returns_bot_ids(self, mock_subprocess: MagicMock) -> None:
|
def test_list_returns_bot_ids(self, mock_list: MagicMock) -> None:
|
||||||
"""Returns list of bot_ids from subprocess output."""
|
"""Returns list of bot_ids from the secrets API."""
|
||||||
mock_result = MagicMock()
|
mock_list.return_value = ["dev_central", "assistant", "scheduler"]
|
||||||
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
|
|
||||||
|
|
||||||
result = tg_config.list_bot_configs()
|
result = tg_config.list_bot_configs()
|
||||||
assert isinstance(result, list)
|
assert isinstance(result, list)
|
||||||
@@ -140,42 +134,28 @@ class TestListBotConfigs:
|
|||||||
assert "assistant" in result
|
assert "assistant" in result
|
||||||
assert "scheduler" in result
|
assert "scheduler" in result
|
||||||
assert len(result) == 3
|
assert len(result) == 3
|
||||||
|
mock_list.assert_called_once_with("telegram")
|
||||||
|
|
||||||
@patch("apps.handlers.config.subprocess")
|
@patch("apps.handlers.config._api_list_secrets")
|
||||||
def test_list_returns_empty_on_failure(self, mock_subprocess: MagicMock) -> None:
|
def test_list_returns_empty_on_failure(self, mock_list: MagicMock) -> None:
|
||||||
"""Returns empty list when subprocess fails."""
|
"""Returns empty list when the secrets API raises."""
|
||||||
mock_result = MagicMock()
|
mock_list.side_effect = RuntimeError("connection failed")
|
||||||
mock_result.returncode = 1
|
|
||||||
mock_result.stdout = ""
|
|
||||||
mock_result.stderr = "error"
|
|
||||||
mock_subprocess.run.return_value = mock_result
|
|
||||||
mock_subprocess.TimeoutExpired = TimeoutError
|
|
||||||
|
|
||||||
result = tg_config.list_bot_configs()
|
result = tg_config.list_bot_configs()
|
||||||
assert result == []
|
assert result == []
|
||||||
|
|
||||||
@patch("apps.handlers.config.subprocess")
|
@patch("apps.handlers.config._api_list_secrets")
|
||||||
def test_list_returns_empty_on_empty_output(self, mock_subprocess: MagicMock) -> None:
|
def test_list_returns_empty_when_no_secrets(self, mock_list: MagicMock) -> None:
|
||||||
"""Returns empty list when subprocess returns empty output."""
|
"""Returns empty list when no secrets exist."""
|
||||||
mock_result = MagicMock()
|
mock_list.return_value = []
|
||||||
mock_result.returncode = 0
|
|
||||||
mock_result.stdout = ""
|
|
||||||
mock_result.stderr = ""
|
|
||||||
mock_subprocess.run.return_value = mock_result
|
|
||||||
mock_subprocess.TimeoutExpired = TimeoutError
|
|
||||||
|
|
||||||
result = tg_config.list_bot_configs()
|
result = tg_config.list_bot_configs()
|
||||||
assert result == []
|
assert result == []
|
||||||
|
|
||||||
@patch("apps.handlers.config.subprocess")
|
@patch("apps.handlers.config._api_list_secrets")
|
||||||
def test_list_returns_empty_on_invalid_json(self, mock_subprocess: MagicMock) -> None:
|
def test_list_returns_empty_on_unexpected_error(self, mock_list: MagicMock) -> None:
|
||||||
"""Returns empty list when subprocess returns invalid JSON."""
|
"""Returns empty list on unexpected exception."""
|
||||||
mock_result = MagicMock()
|
mock_list.side_effect = OSError("disk error")
|
||||||
mock_result.returncode = 0
|
|
||||||
mock_result.stdout = "not valid json"
|
|
||||||
mock_result.stderr = ""
|
|
||||||
mock_subprocess.run.return_value = mock_result
|
|
||||||
mock_subprocess.TimeoutExpired = TimeoutError
|
|
||||||
|
|
||||||
result = tg_config.list_bot_configs()
|
result = tg_config.list_bot_configs()
|
||||||
assert result == []
|
assert result == []
|
||||||
|
|||||||
Reference in New Issue
Block a user