From 2fdbf625680568463cbf33540cfc7b15c4402132 Mon Sep 17 00:00:00 2001 From: AIOSAI Date: Fri, 10 Apr 2026 02:04:56 -0700 Subject: [PATCH] =?UTF-8?q?fix(api):=20S84=20CRITICAL=20security=20fixes?= =?UTF-8?q?=20=E2=80=94=20credential=20exposure,=20file=20permissions,=20v?= =?UTF-8?q?alidation=20alignment?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 3 CRITICAL security fixes: (1) get_cache_stats() no longer exposes raw API keys, (2) .env created with 0o600 perms + dir with 0o700, (3) google_creds.json same restricted permissions. 3 BUG fixes: validation rules aligned between keys.py and provider.py, cleanup return type (0 is success not failure), show_stats() now aggregates across all callers vs show_session() for current session. 290/290 tests pass, seedgo 99%. Co-Authored-By: @api --- src/aipass/api/apps/handlers/auth/env.py | 7 ++- src/aipass/api/apps/handlers/auth/keys.py | 8 +-- src/aipass/api/apps/handlers/google/auth.py | 5 +- .../api/apps/handlers/openrouter/client.py | 3 +- .../api/apps/handlers/usage/aggregation.py | 57 +++++++++++++++++++ src/aipass/api/apps/modules/usage_tracker.py | 22 +++++-- src/aipass/api/tests/test_api_key.py | 14 ++--- src/aipass/api/tests/test_critical_paths.py | 10 ++-- src/aipass/api/tests/test_usage_tracker.py | 47 +++++++-------- 9 files changed, 122 insertions(+), 51 deletions(-) diff --git a/src/aipass/api/apps/handlers/auth/env.py b/src/aipass/api/apps/handlers/auth/env.py index 82658ad7..75000e86 100644 --- a/src/aipass/api/apps/handlers/auth/env.py +++ b/src/aipass/api/apps/handlers/auth/env.py @@ -16,6 +16,7 @@ Functions: """ # Infrastructure +import os from pathlib import Path # Standard library @@ -84,13 +85,17 @@ OPENAI_API_KEY=sk-your-openai-key-here """ try: - # Ensure parent directory exists + # Ensure parent directory exists with restricted permissions (owner-only) env_path.parent.mkdir(parents=True, exist_ok=True) + os.chmod(env_path.parent, 0o700) # Write template with open(env_path, 'w', encoding='utf-8') as f: f.write(env_template) + # Restrict file permissions to owner-read/write only + os.chmod(env_path, 0o600) + # Created .env template logger.info(f"Created .env template at {env_path}") json_handler.log_operation("env_template_created", {"path": str(env_path), "provider": provider}) diff --git a/src/aipass/api/apps/handlers/auth/keys.py b/src/aipass/api/apps/handlers/auth/keys.py index 076e4bbc..bfd3d947 100644 --- a/src/aipass/api/apps/handlers/auth/keys.py +++ b/src/aipass/api/apps/handlers/auth/keys.py @@ -42,16 +42,16 @@ API_JSON_DIR = API_ROOT / "api_json" # Provider validation rules (embedded - no config dependency for core validation) VALIDATION_RULES = { "openrouter": { - "prefix": "sk-or-", - "min_length": 20 + "prefix": "sk-or-v1-", + "min_length": 40 }, "openai": { "prefix": "sk-", - "min_length": 20 + "min_length": 40 }, "anthropic": { "prefix": "sk-ant-", - "min_length": 20 + "min_length": 40 }, # Generic fallback "generic": { diff --git a/src/aipass/api/apps/handlers/google/auth.py b/src/aipass/api/apps/handlers/google/auth.py index 71587bcc..745c008f 100644 --- a/src/aipass/api/apps/handlers/google/auth.py +++ b/src/aipass/api/apps/handlers/google/auth.py @@ -20,6 +20,7 @@ This is pure auth plumbing — no business logic. Consumers get authenticated credentials, they decide what to do with them. """ +import os from pathlib import Path from typing import Optional @@ -245,7 +246,9 @@ def validate_credentials(scopes: Optional[list] = None) -> bool: def _save_credentials(creds: "Credentials") -> None: - """Save credentials to the standard secrets path.""" + """Save credentials to the standard secrets path with restricted permissions.""" SECRETS_DIR.mkdir(parents=True, exist_ok=True) + os.chmod(SECRETS_DIR, 0o700) with open(CREDS_PATH, "w", encoding="utf-8") as f: f.write(creds.to_json()) + os.chmod(CREDS_PATH, 0o600) diff --git a/src/aipass/api/apps/handlers/openrouter/client.py b/src/aipass/api/apps/handlers/openrouter/client.py index 34960baf..e93314ed 100644 --- a/src/aipass/api/apps/handlers/openrouter/client.py +++ b/src/aipass/api/apps/handlers/openrouter/client.py @@ -367,10 +367,9 @@ def get_cache_stats() -> Dict[str, Any]: Get statistics about the client cache. Returns: - Dict with cache size and keys + Dict with cache size info (keys are never exposed — they are raw API keys) """ return { "cached_clients": len(_client_cache), "max_cache_size": MAX_CACHED_CLIENTS, - "cache_keys": list(_client_cache.keys()) } diff --git a/src/aipass/api/apps/handlers/usage/aggregation.py b/src/aipass/api/apps/handlers/usage/aggregation.py index a48980ae..2f9bca31 100644 --- a/src/aipass/api/apps/handlers/usage/aggregation.py +++ b/src/aipass/api/apps/handlers/usage/aggregation.py @@ -47,6 +47,63 @@ API_JSON_DIR = Path(__file__).resolve().parent.parent.parent.parent / "api_json" # AGGREGATION FUNCTIONS # ============================================= +def get_overall_stats() -> Dict[str, Any]: + """ + Aggregate usage statistics across all callers. + + Returns: + Dict with total_requests, total_cost, total_tokens, callers (count), + models_used (set of model names). + Returns empty dict {} if no data found. + """ + try: + data_path = API_JSON_DIR / DATA_FILE + if not data_path.exists(): + logger.info(f"[{MODULE_NAME}] No usage data file found") + return {} + + with open(data_path, 'r', encoding='utf-8') as f: + data = json.load(f) + + if not data or "data" not in data: + logger.info(f"[{MODULE_NAME}] No usage data available") + return {} + + usage_by_caller = data["data"].get("usage_by_caller", {}) + + if not usage_by_caller: + logger.info(f"[{MODULE_NAME}] No caller usage data found") + return {} + + total_requests = 0 + total_cost = 0.0 + total_tokens = 0 + models_used: set = set() + + for caller_data in usage_by_caller.values(): + total_requests += caller_data.get("requests", 0) + total_cost += caller_data.get("total_cost", 0.0) + total_tokens += caller_data.get("total_tokens", 0) + for model in caller_data.get("models_used", []): + models_used.add(model) + + result = { + "total_requests": total_requests, + "total_cost": total_cost, + "total_tokens": total_tokens, + "callers": len(usage_by_caller), + "models_used": sorted(models_used), + } + + logger.info(f"[{MODULE_NAME}] Overall stats: {total_requests} requests across {len(usage_by_caller)} callers") + json_handler.log_operation("get_overall_stats", {"total_requests": total_requests}) + return result + + except Exception as e: + logger.error(f"[{MODULE_NAME}] Failed to get overall stats: {e}") + return {} + + def get_caller_usage(caller: str) -> Dict[str, Any]: """ Calculate usage statistics for specific caller diff --git a/src/aipass/api/apps/modules/usage_tracker.py b/src/aipass/api/apps/modules/usage_tracker.py index e4af9141..56b174de 100644 --- a/src/aipass/api/apps/modules/usage_tracker.py +++ b/src/aipass/api/apps/modules/usage_tracker.py @@ -195,17 +195,20 @@ def track_usage(args: List[str]): def show_stats(): - """Orchestrate statistics display workflow""" + """Orchestrate overall statistics display workflow (aggregate across all callers)""" header("Usage Statistics") console.print() - # Call handler for session summary - stats = aggregation.get_session_summary() + stats = aggregation.get_overall_stats() if stats: console.print(f" Total Requests: {stats.get('total_requests', 0)}") console.print(f" Total Cost: ${stats.get('total_cost', 0.0):.6f}") console.print(f" Total Tokens: {stats.get('total_tokens', 0)}") + console.print(f" Callers: {stats.get('callers', 0)}") + models = stats.get('models_used', []) + if models: + console.print(f" Models Used: {', '.join(models)}") else: warning("No usage data available") @@ -259,8 +262,15 @@ def cleanup_data(args: List[str]): # Navigate: usage_tracker.py -> modules/ -> apps/ -> api/ API_JSON_DIR = Path(__file__).resolve().parent.parent.parent / "api_json" data_path = API_JSON_DIR / "usage_tracker_data.json" - if cleanup.cleanup_old_data(data_path, days): - success(f"Cleaned up data older than {days} days") + + if not data_path.exists(): + warning("No usage data file found — nothing to clean") + return + + removed = cleanup.cleanup_old_data(data_path, days) + + if removed > 0: + success(f"Cleaned up {removed} entries older than {days} days") # Fire trigger event try: @@ -269,7 +279,7 @@ def cleanup_data(args: List[str]): except ImportError: logger.warning("Trigger module not available — skipping event fire") else: - error("Cleanup failed") + success(f"Nothing to clean — no entries older than {days} days") if __name__ == "__main__": diff --git a/src/aipass/api/tests/test_api_key.py b/src/aipass/api/tests/test_api_key.py index d3dafa66..ebd70b26 100644 --- a/src/aipass/api/tests/test_api_key.py +++ b/src/aipass/api/tests/test_api_key.py @@ -660,22 +660,22 @@ class TestGetValidationRulesAuthKeys: """Tests for auth.keys.get_validation_rules().""" def test_openrouter_rules(self): - """openrouter should have prefix 'sk-or-' and min_length 20.""" + """openrouter should have prefix 'sk-or-v1-' and min_length 40.""" rules = auth_keys.get_validation_rules("openrouter") - assert rules["prefix"] == "sk-or-" - assert rules["min_length"] == 20 + assert rules["prefix"] == "sk-or-v1-" + assert rules["min_length"] == 40 def test_openai_rules(self): - """openai should have prefix 'sk-' and min_length 20.""" + """openai should have prefix 'sk-' and min_length 40.""" rules = auth_keys.get_validation_rules("openai") assert rules["prefix"] == "sk-" - assert rules["min_length"] == 20 + assert rules["min_length"] == 40 def test_anthropic_rules(self): - """anthropic should have prefix 'sk-ant-' and min_length 20.""" + """anthropic should have prefix 'sk-ant-' and min_length 40.""" rules = auth_keys.get_validation_rules("anthropic") assert rules["prefix"] == "sk-ant-" - assert rules["min_length"] == 20 + assert rules["min_length"] == 40 def test_unknown_provider_falls_back_to_generic(self): """Unknown provider should fall back to generic rules.""" diff --git a/src/aipass/api/tests/test_critical_paths.py b/src/aipass/api/tests/test_critical_paths.py index 07e0c516..23f54fa9 100644 --- a/src/aipass/api/tests/test_critical_paths.py +++ b/src/aipass/api/tests/test_critical_paths.py @@ -45,7 +45,7 @@ class TestGetApiKey: "config": { "providers": { "openrouter": { - "api_key": "sk-or-v1-valid-key-that-is-long-enough" + "api_key": "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" } } } @@ -56,7 +56,7 @@ class TestGetApiKey: result = get_api_key("openrouter") - assert result == "sk-or-v1-valid-key-that-is-long-enough" + assert result == "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" mock_jh.log_operation.assert_called_once() @patch("aipass.api.apps.handlers.auth.keys.json_handler") @@ -76,7 +76,7 @@ class TestGetApiKey: secrets_dir.mkdir(parents=True) env_file = secrets_dir / ".env" env_file.write_text( - "OPENROUTER_API_KEY=sk-or-v1-secret-key-long-enough-here\n", + "OPENROUTER_API_KEY=sk-or-v1-a1b2c3d4e5f6a1b2c3d4e5f6a1b2c3d4\n", encoding="utf-8", ) @@ -85,7 +85,7 @@ class TestGetApiKey: mock_path_cls.home.return_value = tmp_path result = get_api_key("openrouter") - assert result == "sk-or-v1-secret-key-long-enough-here" + assert result == "sk-or-v1-a1b2c3d4e5f6a1b2c3d4e5f6a1b2c3d4" @patch("aipass.api.apps.handlers.auth.keys.json_handler") @patch("aipass.api.apps.handlers.auth.keys.API_JSON_DIR") @@ -146,7 +146,7 @@ class TestValidateKey: """Valid openrouter key with correct prefix and length passes.""" from aipass.api.apps.handlers.auth.keys import validate_key - key = "sk-or-v1-abcdefghijklmnopqrst" + key = "sk-or-v1-abcdefghijklmnopqrstuvwxyz0123456" assert validate_key(key, "openrouter") is True def test_key_too_short(self): diff --git a/src/aipass/api/tests/test_usage_tracker.py b/src/aipass/api/tests/test_usage_tracker.py index 7522abf9..05a317d1 100644 --- a/src/aipass/api/tests/test_usage_tracker.py +++ b/src/aipass/api/tests/test_usage_tracker.py @@ -196,10 +196,12 @@ def test_show_stats_with_data(mock_agg, mock_header, mock_console, mock_warning) """show_stats prints stats when aggregation returns data.""" from aipass.api.apps.modules import usage_tracker - mock_agg.get_session_summary.return_value = { + mock_agg.get_overall_stats.return_value = { "total_requests": 42, "total_cost": 0.123456, "total_tokens": 9001, + "callers": 3, + "models_used": ["anthropic/claude-3.5-sonnet"], } usage_tracker.show_stats() @@ -221,7 +223,7 @@ def test_show_stats_no_data(mock_agg, mock_header, mock_console, mock_warning): """show_stats shows warning when no data available.""" from aipass.api.apps.modules import usage_tracker - mock_agg.get_session_summary.return_value = {} + mock_agg.get_overall_stats.return_value = {} usage_tracker.show_stats() @@ -360,45 +362,40 @@ def test_show_caller_usage_no_args(mock_agg, mock_header, mock_console, mock_err @patch(f"{PATCH_ROOT}.header") @patch(f"{PATCH_ROOT}.cleanup") def test_cleanup_success(mock_cleanup_handler, mock_header, mock_console, mock_success, mock_error): - """cleanup_data calls success() when handler returns truthy.""" + """cleanup_data calls success() with count when handler returns > 0.""" from aipass.api.apps.modules import usage_tracker - mock_cleanup_handler.cleanup_old_data.return_value = True + mock_cleanup_handler.cleanup_old_data.return_value = 5 + # The data_path.exists() check needs to pass with patch(f"{PATCH_ROOT}.Path") as mock_path_cls: - mock_path_cls.__file__ = MagicMock() - # Let Path(__file__).resolve().parent chain work - mock_resolved = MagicMock() - mock_path_cls.return_value.resolve.return_value.parent.parent.parent.__truediv__ = MagicMock() + mock_data_path = MagicMock() + mock_data_path.exists.return_value = True + mock_path_cls.return_value.resolve.return_value.parent.parent.parent.__truediv__.return_value.__truediv__.return_value = mock_data_path - # Simpler approach: just let the real Path work -- it resolves against - # the actual source file, but cleanup_old_data is mocked anyway. - pass - - # Call directly without patching Path -- cleanup handler is mocked + # Call directly -- cleanup handler is mocked, use real Path for API_JSON_DIR resolution usage_tracker.cleanup_data(["45"]) mock_cleanup_handler.cleanup_old_data.assert_called_once() - mock_success.assert_called_once_with("Cleaned up data older than 45 days") + mock_success.assert_called_once_with("Cleaned up 5 entries older than 45 days") mock_error.assert_not_called() -@patch(f"{PATCH_ROOT}.error") +@patch(f"{PATCH_ROOT}.warning") @patch(f"{PATCH_ROOT}.success") @patch(f"{PATCH_ROOT}.console") @patch(f"{PATCH_ROOT}.header") @patch(f"{PATCH_ROOT}.cleanup") -def test_cleanup_failure(mock_cleanup_handler, mock_header, mock_console, mock_success, mock_error): - """cleanup_data calls error() when handler returns falsy.""" +def test_cleanup_nothing_to_clean(mock_cleanup_handler, mock_header, mock_console, mock_success, mock_warning): + """cleanup_data calls success() with 'nothing to clean' when handler returns 0.""" from aipass.api.apps.modules import usage_tracker - mock_cleanup_handler.cleanup_old_data.return_value = False + mock_cleanup_handler.cleanup_old_data.return_value = 0 usage_tracker.cleanup_data(["30"]) mock_cleanup_handler.cleanup_old_data.assert_called_once() - mock_error.assert_called_once_with("Cleanup failed") - mock_success.assert_not_called() + mock_success.assert_called_once_with("Nothing to clean — no entries older than 30 days") @patch(f"{PATCH_ROOT}.error") @@ -410,7 +407,7 @@ def test_cleanup_default_30_days(mock_cleanup_handler, mock_header, mock_console """cleanup_data defaults to 30 days when no args provided.""" from aipass.api.apps.modules import usage_tracker - mock_cleanup_handler.cleanup_old_data.return_value = True + mock_cleanup_handler.cleanup_old_data.return_value = 3 usage_tracker.cleanup_data([]) @@ -419,7 +416,7 @@ def test_cleanup_default_30_days(mock_cleanup_handler, mock_header, mock_console # Verify cleanup_old_data was called with days=30 args, kwargs = mock_cleanup_handler.cleanup_old_data.call_args assert args[1] == 30 - mock_success.assert_called_once_with("Cleaned up data older than 30 days") + mock_success.assert_called_once_with("Cleaned up 3 entries older than 30 days") @patch(f"{PATCH_ROOT}.error") @@ -431,14 +428,14 @@ def test_cleanup_custom_days(mock_cleanup_handler, mock_header, mock_console, mo """cleanup_data parses custom days from args.""" from aipass.api.apps.modules import usage_tracker - mock_cleanup_handler.cleanup_old_data.return_value = True + mock_cleanup_handler.cleanup_old_data.return_value = 7 usage_tracker.cleanup_data(["90"]) mock_header.assert_called_once_with("Cleanup Old Data (retain 90 days)") args, kwargs = mock_cleanup_handler.cleanup_old_data.call_args assert args[1] == 90 - mock_success.assert_called_once_with("Cleaned up data older than 90 days") + mock_success.assert_called_once_with("Cleaned up 7 entries older than 90 days") # ============================================= @@ -454,7 +451,7 @@ def test_handle_command_propagates_exception(mock_console, mock_header, mock_jh, """handle_command re-raises exceptions from downstream handlers.""" from aipass.api.apps.modules import usage_tracker - mock_agg.get_session_summary.side_effect = RuntimeError("handler failed") + mock_agg.get_overall_stats.side_effect = RuntimeError("handler failed") with pytest.raises(RuntimeError, match="handler failed"): usage_tracker.handle_command("stats", [])