Merge pull request #220 from AIOSAI/fix/api-s84-security-fixes

fix(api): S84 CRITICAL security fixes

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})

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": {

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)

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())
     }

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

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__":

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."""

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):

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", [])