diff --git a/scripts/cli_tool_bridge.py b/scripts/cli_tool_bridge.py new file mode 100644 index 0000000..e912805 --- /dev/null +++ b/scripts/cli_tool_bridge.py @@ -0,0 +1,65 @@ +import sys +import json +import logging +import os + +# Add project root to sys.path so we can import api_hook_client +sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))) + +try: + from api_hook_client import ApiHookClient +except ImportError: + # Fallback for if we are running from root or other locations + sys.path.append(os.path.abspath(os.path.dirname(__file__))) + from api_hook_client import ApiHookClient + +def main(): + # Setup basic logging to stderr so it doesn't interfere with stdout JSON + logging.basicConfig(level=logging.ERROR, stream=sys.stderr) + + try: + # 1. Read JSON from sys.stdin + input_data = sys.stdin.read() + if not input_data: + return + + hook_input = json.loads(input_data) + + # 2. Extract 'tool_name' and 'tool_input' + tool_name = hook_input.get('tool_name') + tool_args = hook_input.get('tool_input', {}) + + # 3. Use 'ApiHookClient' (assuming GUI is on http://127.0.0.1:8999) + client = ApiHookClient(base_url="http://127.0.0.1:8999") + + try: + # 5. Request confirmation + # This is a blocking call that waits for the user in the GUI + response = client.request_confirmation(tool_name, tool_args) + + if response and response.get('approved') is True: + # 5. Print 'allow' decision + print(json.dumps({"decision": "allow"})) + else: + # 6. Print 'deny' decision + print(json.dumps({ + "decision": "deny", + "reason": "User rejected tool execution." + })) + + except Exception as e: + # 7. Handle cases where hook server is not reachable + print(json.dumps({ + "decision": "deny", + "reason": f"Hook server unreachable or error occurred: {str(e)}" + })) + + except Exception as e: + # Fallback for unexpected parsing errors + print(json.dumps({ + "decision": "deny", + "reason": f"Internal bridge error: {str(e)}" + })) + +if __name__ == "__main__": + main() diff --git a/tests/test_cli_tool_bridge.py b/tests/test_cli_tool_bridge.py new file mode 100644 index 0000000..e7ecc84 --- /dev/null +++ b/tests/test_cli_tool_bridge.py @@ -0,0 +1,74 @@ +import unittest +from unittest.mock import patch, MagicMock +import io +import json +import sys +import os + +# Add project root to sys.path +sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))) + +# Import after path fix +from scripts.cli_tool_bridge import main + +class TestCliToolBridge(unittest.TestCase): + def setUp(self): + self.tool_call = { + 'tool_name': 'read_file', + 'tool_input': {'path': 'test.txt'} + } + + @patch('sys.stdin', new_callable=io.StringIO) + @patch('sys.stdout', new_callable=io.StringIO) + @patch('api_hook_client.ApiHookClient.request_confirmation') + def test_allow_decision(self, mock_request, mock_stdout, mock_stdin): + # 1. Mock stdin with a JSON string tool call + mock_stdin.write(json.dumps(self.tool_call)) + mock_stdin.seek(0) + + # 2. Mock ApiHookClient to return approved + mock_request.return_value = {'approved': True} + + # Run main + main() + + # 3. Capture stdout and assert allow + output = json.loads(mock_stdout.getvalue().strip()) + self.assertEqual(output.get('decision'), 'allow') + + @patch('sys.stdin', new_callable=io.StringIO) + @patch('sys.stdout', new_callable=io.StringIO) + @patch('api_hook_client.ApiHookClient.request_confirmation') + def test_deny_decision(self, mock_request, mock_stdout, mock_stdin): + # Mock stdin + mock_stdin.write(json.dumps(self.tool_call)) + mock_stdin.seek(0) + + # 4. Mock ApiHookClient to return denied + mock_request.return_value = {'approved': False} + + main() + + # Assert deny + output = json.loads(mock_stdout.getvalue().strip()) + self.assertEqual(output.get('decision'), 'deny') + + @patch('sys.stdin', new_callable=io.StringIO) + @patch('sys.stdout', new_callable=io.StringIO) + @patch('api_hook_client.ApiHookClient.request_confirmation') + def test_unreachable_hook_server(self, mock_request, mock_stdout, mock_stdin): + # Mock stdin + mock_stdin.write(json.dumps(self.tool_call)) + mock_stdin.seek(0) + + # 5. Test case where hook server is unreachable (exception) + mock_request.side_effect = Exception("Connection refused") + + main() + + # Assert deny on error + output = json.loads(mock_stdout.getvalue().strip()) + self.assertEqual(output.get('decision'), 'deny') + +if __name__ == '__main__': + unittest.main()