🛡️ Sentinel: Fix info leak in login_required decorator

auth/utils.py

@@ -1,14 +1,20 @@
-import bcrypt
-import jwt
-import os
-import secrets
+"""Authentication and session management utilities."""
+
 from datetime import datetime, timedelta
 from functools import wraps
+import logging
+import os
+import secrets
+from typing import Tuple, Optional
+
+import bcrypt
+import jwt
 from flask import request, jsonify, g
 from flask_jwt_extended import get_jwt_identity, verify_jwt_in_request
 from cache_db.redis_client import redis_client
 from cache_db.models import User, RefreshToken
-from typing import Tuple, Optional
+
+logger = logging.getLogger(__name__)
 
 
 class PasswordUtils:
@@ -129,7 +135,8 @@ def decorated_function(*args, **kwargs):
             g.user = user_data
             return f(*args, **kwargs)
         except Exception as e:
-            return jsonify({'error': 'Unauthorized', 'details': str(e)}), 401
+            logger.error("Authentication check failed: %s", e)
+            return jsonify({'error': 'Unauthorized'}), 401
     return decorated_function
 
 

tests/test_auth.py

@@ -0,0 +1,40 @@
+import sys
+from unittest.mock import MagicMock, patch
+from flask import Flask
+
+# Mock cache_db modules before importing auth.utils
+mock_redis = MagicMock()
+mock_models = MagicMock()
+sys.modules['cache_db'] = MagicMock()
+sys.modules['cache_db.redis_client'] = MagicMock(redis_client=mock_redis)
+sys.modules['cache_db.models'] = MagicMock(User=mock_models.User, RefreshToken=mock_models.RefreshToken)
+
+from auth.utils import PasswordUtils, JWTUtils, login_required, admin_required, verify_required
+
+
+def test_password_utils():
+    hashed = PasswordUtils.hash_password("securepassword123")
+    assert PasswordUtils.verify_password("securepassword123", hashed)
+    assert not PasswordUtils.verify_password("wrongpassword", hashed)
+
+
+def test_login_required_does_not_leak_exception_details():
+    app = Flask(__name__)
+    app.config["SECRET_KEY"] = "test-secret"
+
+    @app.route("/protected")
+    @login_required
+    def protected():
+        return "ok"
+
+    with app.test_client() as client:
+        # Simulate a request where JWT verification raises an exception with sensitive internal details
+        with patch("auth.utils.verify_jwt_in_request", side_effect=RuntimeError("Database host 10.0.0.5 connection failed")):
+            response = client.get("/protected")
+            assert response.status_code == 401
+            json_data = response.get_json()
+            assert json_data == {"error": "Unauthorized"}
+            # Ensure sensitive internal exception string is NOT in response
+            assert "details" not in json_data
+            assert "10.0.0.5" not in response.get_data(as_text=True)
+            assert "Database" not in response.get_data(as_text=True)

tests/test_automation.py

@@ -0,0 +1,11 @@
+from automation.engine import AutomationEngine, AutomationDefinition, TriggerEvent, RunStatus
+
+
+def test_automation_engine_basic_flow():
+    engine = AutomationEngine()
+    engine.register_fn("greet", lambda inp: f"Hello, {inp.payload}!")
+    engine.define(AutomationDefinition(name="hello", triggers=["user.request"], steps=["greet"]))
+    results = engine.trigger_type("user.request", payload="world")
+    assert len(results) == 1
+    assert results[0].status == RunStatus.SUCCESS
+    assert results[0].outputs[0].result == "Hello, world!"

tests/test_handoff.py

@@ -0,0 +1,18 @@
+from handoff.handoff import HandoffManager, HandoffStatus
+
+
+def test_handoff_manager_flow():
+    manager = HandoffManager()
+    context = manager.initiate_handoff(
+        task_id="task:100",
+        source_agent="agent-a",
+        target_agent="agent-b",
+        task_state={"step": 1},
+    )
+    assert context.handoff_id is not None
+    assert manager.prepare_handoff(context.handoff_id) is True
+    assert manager.transmit_handoff(context.handoff_id) is True
+    assert manager.receive_handoff(context.handoff_id) is True
+    assert manager.accept_handoff(context.handoff_id) is True
+    assert manager.complete_handoff(context.handoff_id) is True
+    assert manager.get_handoff_status(context.handoff_id) == HandoffStatus.COMPLETED