security

backend/tests/integration/debug_endpoints.py

@@ -1,169 +0,0 @@
-#!/usr/bin/env python3
-"""Debug script to check which endpoints are working."""
-
-import requests
-import json
-from urllib.parse import urljoin
-
-
-def test_endpoint(url, method="GET", headers=None, data=None, timeout=5):
-    """Test a single endpoint and return detailed results."""
-    try:
-        if method.upper() == "GET":
-            response = requests.get(url, headers=headers, timeout=timeout)
-        elif method.upper() == "POST":
-            response = requests.post(url, headers=headers, json=data, timeout=timeout)
-
-        return {
-            "status": "SUCCESS",
-            "status_code": response.status_code,
-            "response": response.text[:500],  # Limit response length
-            "headers": dict(response.headers),
-        }
-    except requests.exceptions.ConnectionError:
-        return {"status": "CONNECTION_ERROR", "error": "Cannot connect to server"}
-    except requests.exceptions.Timeout:
-        return {"status": "TIMEOUT", "error": "Request timed out"}
-    except Exception as e:
-        return {"status": "ERROR", "error": str(e)}
-
-
-def check_all_endpoints():
-    """Check all possible endpoints systematically."""
-
-    print("🔍 Debugging Endpoint Availability")
-    print("=" * 60)
-
-    # Define test cases
-    test_cases = [
-        # FastAPI Server (Port 8000)
-        {
-            "name": "FastAPI Health",
-            "url": "http://localhost:8000/health",
-            "method": "GET",
-        },
-        {"name": "FastAPI Root", "url": "http://localhost:8000/", "method": "GET"},
-        {
-            "name": "FastAPI API Root",
-            "url": "http://localhost:8000/api/",
-            "method": "GET",
-        },
-        # MCP Server (Port 8001)
-        {"name": "MCP Root", "url": "http://localhost:8001/", "method": "GET"},
-        {"name": "MCP Endpoint", "url": "http://localhost:8001/mcp", "method": "GET"},
-        {
-            "name": "MCP Health (if exists)",
-            "url": "http://localhost:8001/health",
-            "method": "GET",
-        },
-        # MCP Protocol Tests
-        {
-            "name": "MCP Initialize",
-            "url": "http://localhost:8001/mcp",
-            "method": "POST",
-            "headers": {
-                "Authorization": "Bearer local-dev-token",
-                "Content-Type": "application/json",
-                "Accept": "application/json, text/event-stream",
-            },
-            "data": {
-                "jsonrpc": "2.0",
-                "id": 1,
-                "method": "initialize",
-                "params": {
-                    "protocolVersion": "2024-11-05",
-                    "capabilities": {},
-                    "clientInfo": {"name": "debug-client", "version": "1.0.0"},
-                },
-            },
-        },
-        {
-            "name": "MCP Tools List",
-            "url": "http://localhost:8001/mcp",
-            "method": "POST",
-            "headers": {
-                "Authorization": "Bearer local-dev-token",
-                "Content-Type": "application/json",
-                "Accept": "application/json, text/event-stream",
-            },
-            "data": {"jsonrpc": "2.0", "id": 2, "method": "tools/list"},
-        },
-    ]
-
-    results = {}
-
-    for test in test_cases:
-        print(f"\n🧪 Testing: {test['name']}")
-        print(f"   URL: {test['url']}")
-        print(f"   Method: {test['method']}")
-
-        result = test_endpoint(
-            url=test["url"],
-            method=test["method"],
-            headers=test.get("headers"),
-            data=test.get("data"),
-        )
-
-        results[test["name"]] = result
-
-        if result["status"] == "SUCCESS":
-            print(f"   ✅ Status: {result['status_code']}")
-            if result["response"]:
-                # Try to pretty print JSON
-                try:
-                    json_resp = json.loads(result["response"])
-                    print(f"   📄 Response: {json.dumps(json_resp, indent=2)[:200]}...")
-                except:
-                    print(f"   📄 Response: {result['response'][:100]}...")
-        else:
-            print(f"   ❌ {result['status']}: {result.get('error', 'Unknown error')}")
-
-    print("\n" + "=" * 60)
-    print("📊 SUMMARY")
-    print("=" * 60)
-
-    working = []
-    not_working = []
-
-    for name, result in results.items():
-        if result["status"] == "SUCCESS" and result.get("status_code", 0) < 400:
-            working.append(name)
-        else:
-            not_working.append(name)
-
-    print(f"\n✅ WORKING ({len(working)}):")
-    for name in working:
-        status_code = results[name].get("status_code", "N/A")
-        print(f"   - {name} (HTTP {status_code})")
-
-    print(f"\n❌ NOT WORKING ({len(not_working)}):")
-    for name in not_working:
-        result = results[name]
-        if result["status"] == "SUCCESS":
-            print(f"   - {name} (HTTP {result.get('status_code', 'N/A')})")
-        else:
-            print(f"   - {name} ({result['status']})")
-
-    print(f"\n🔧 RECOMMENDATIONS:")
-
-    # Check if any server is running
-    fastapi_working = any("FastAPI" in name for name in working)
-    mcp_working = any("MCP" in name for name in working)
-
-    if not fastapi_working:
-        print("   - Start FastAPI server: cd backend && python main.py")
-
-    if not mcp_working:
-        print(
-            "   - Start MCP server: cd backend && MCP_TRANSPORT=http MCP_PORT=8001 python -m src.mcp.server"
-        )
-
-    if not working:
-        print("   - Check if any servers are running: netstat -ano | findstr :8000")
-        print("   - Check if any servers are running: netstat -ano | findstr :8001")
-
-    return results
-
-
-if __name__ == "__main__":
-    check_all_endpoints()

debug_endpoints.py

@@ -1,170 +0,0 @@
-#!/usr/bin/env python3
-"""Debug script to check which endpoints are working."""
-
-import platform
-import requests
-import json
-from urllib.parse import urljoin
-
-
-def test_endpoint(url, method="GET", headers=None, data=None, timeout=5):
-    """Test a single endpoint and return detailed results."""
-    try:
-        if method.upper() == "GET":
-            response = requests.get(url, headers=headers, timeout=timeout)
-        elif method.upper() == "POST":
-            response = requests.post(url, headers=headers, json=data, timeout=timeout)
-
-        return {
-            "status": "SUCCESS",
-            "status_code": response.status_code,
-            "response": response.text[:500],  # Limit response length
-            "headers": dict(response.headers),
-        }
-    except requests.exceptions.ConnectionError:
-        return {"status": "CONNECTION_ERROR", "error": "Cannot connect to server"}
-    except requests.exceptions.Timeout:
-        return {"status": "TIMEOUT", "error": "Request timed out"}
-    except Exception as e:
-        return {"status": "ERROR", "error": str(e)}
-
-
-def check_all_endpoints():
-    """Check all possible endpoints systematically."""
-
-    print("🔍 Debugging Endpoint Availability")
-    print("=" * 60)
-
-    # Define test cases
-    test_cases = [
-        # FastAPI Server (Port 8000)
-        {
-            "name": "FastAPI Health",
-            "url": "http://localhost:8000/health",
-            "method": "GET",
-        },
-        {"name": "FastAPI Root", "url": "http://localhost:8000/", "method": "GET"},
-        {
-            "name": "FastAPI API Root",
-            "url": "http://localhost:8000/api/",
-            "method": "GET",
-        },
-        # MCP Server (Port 8001)
-        {"name": "MCP Root", "url": "http://localhost:8001/", "method": "GET"},
-        {"name": "MCP Endpoint", "url": "http://localhost:8001/mcp", "method": "GET"},
-        {
-            "name": "MCP Health (if exists)",
-            "url": "http://localhost:8001/health",
-            "method": "GET",
-        },
-        # MCP Protocol Tests
-        {
-            "name": "MCP Initialize",
-            "url": "http://localhost:8001/mcp",
-            "method": "POST",
-            "headers": {
-                "Authorization": "Bearer local-dev-token",
-                "Content-Type": "application/json",
-                "Accept": "application/json, text/event-stream",
-            },
-            "data": {
-                "jsonrpc": "2.0",
-                "id": 1,
-                "method": "initialize",
-                "params": {
-                    "protocolVersion": "2024-11-05",
-                    "capabilities": {},
-                    "clientInfo": {"name": "debug-client", "version": "1.0.0"},
-                },
-            },
-        },
-        {
-            "name": "MCP Tools List",
-            "url": "http://localhost:8001/mcp",
-            "method": "POST",
-            "headers": {
-                "Authorization": "Bearer local-dev-token",
-                "Content-Type": "application/json",
-                "Accept": "application/json, text/event-stream",
-            },
-            "data": {"jsonrpc": "2.0", "id": 2, "method": "tools/list"},
-        },
-    ]
-
-    results = {}
-
-    for test in test_cases:
-        print(f"\n🧪 Testing: {test['name']}")
-        print(f"   URL: {test['url']}")
-        print(f"   Method: {test['method']}")
-
-        result = test_endpoint(
-            url=test["url"],
-            method=test["method"],
-            headers=test.get("headers"),
-            data=test.get("data"),
-        )
-
-        results[test["name"]] = result
-
-        if result["status"] == "SUCCESS":
-            print(f"   ✅ Status: {result['status_code']}")
-            if result["response"]:
-                # Try to pretty print JSON
-                try:
-                    json_resp = json.loads(result["response"])
-                    print(f"   📄 Response: {json.dumps(json_resp, indent=2)[:200]}...")
-                except:
-                    print(f"   📄 Response: {result['response'][:100]}...")
-        else:
-            print(f"   ❌ {result['status']}: {result.get('error', 'Unknown error')}")
-
-    print("\n" + "=" * 60)
-    print("📊 SUMMARY")
-    print("=" * 60)
-
-    working = []
-    not_working = []
-
-    for name, result in results.items():
-        if result["status"] == "SUCCESS" and result.get("status_code", 0) < 400:
-            working.append(name)
-        else:
-            not_working.append(name)
-
-    print(f"\n✅ WORKING ({len(working)}):")
-    for name in working:
-        status_code = results[name].get("status_code", "N/A")
-        print(f"   - {name} (HTTP {status_code})")
-
-    print(f"\n❌ NOT WORKING ({len(not_working)}):")
-    for name in not_working:
-        result = results[name]
-        if result["status"] == "SUCCESS":
-            print(f"   - {name} (HTTP {result.get('status_code', 'N/A')})")
-        else:
-            print(f"   - {name} ({result['status']})")
-
-    print(f"\n🔧 RECOMMENDATIONS:")
-
-    # Check if any server is running
-    fastapi_working = any("FastAPI" in name for name in working)
-    mcp_working = any("MCP" in name for name in working)
-
-    if not fastapi_working:
-        print("   - Start FastAPI server: cd backend && python main.py")
-
-    if not mcp_working:
-        print(
-            "   - Start MCP server: cd backend && MCP_TRANSPORT=http MCP_PORT=8001 python -m src.mcp.server"
-        )
-
-    if not working:
-        print("   - Check if any servers are running: netstat -ano | findstr :8000")
-        print("   - Check if any servers are running: netstat -ano | findstr :8001")
-
-    return results
-
-
-if __name__ == "__main__":
-    check_all_endpoints()

debug_mcp_endpoint.py

@@ -1,45 +0,0 @@
-
-import asyncio
-import logging
-from fastapi.testclient import TestClient
-from backend.src.api.main import app
-
-# Configure logging
-logging.basicConfig(level=logging.DEBUG)
-
-def debug_mcp():
-    client = TestClient(app)
-    
-    print("--- Test 1: GET /mcp (No Auth) ---")
-    try:
-        response = client.get("/mcp")
-        print(f"Status: {response.status_code}")
-        print(f"Body: {response.text}")
-    except Exception as e:
-        print(f"Crash: {e}")
-
-    print("\n--- Test 2: POST /mcp (No Auth) - List Tools ---")
-    # JSON-RPC payload
-    payload = {
-        "jsonrpc": "2.0",
-        "method": "tools/list",
-        "id": 1
-    }
-    try:
-        response = client.post("/mcp", json=payload)
-        print(f"Status: {response.status_code}")
-        print(f"Body: {response.text}")
-    except Exception as e:
-        print(f"Crash: {e}")
-
-    print("\n--- Test 3: POST /mcp (With Auth) - List Tools ---")
-    try:
-        # Assuming local-dev-token is valid
-        response = client.post("/mcp", json=payload, headers={"Authorization": "Bearer local-dev-token"})
-        print(f"Status: {response.status_code}")
-        print(f"Body: {response.text}")
-    except Exception as e:
-        print(f"Crash: {e}")
-
-if __name__ == "__main__":
-    debug_mcp()

generate_jwt_token.py

@@ -1,38 +0,0 @@
-#!/usr/bin/env python3
-"""Generate a JWT token for testing MCP HTTP transport."""
-
-import sys
-import os
-
-# Add backend to path
-sys.path.insert(0, "./backend")
-
-from backend.src.services.auth import AuthService
-from backend.src.services.config import get_config
-
-
-def generate_token(user_id="local-dev"):
-    """Generate a JWT token for the specified user."""
-    try:
-        config = get_config()
-        auth_service = AuthService(config=config)
-
-        # Generate JWT token
-        token = auth_service.create_jwt(user_id)
-
-        print(f"✅ Generated JWT token for user '{user_id}':")
-        print(f"Bearer {token}")
-        print(f"\n📋 Copy this to your mcp.json:")
-        print(f'"Authorization": "Bearer {token}"')
-
-        return token
-
-    except Exception as e:
-        print(f"❌ Error generating token: {e}")
-        print("💡 Make sure JWT_SECRET_KEY is set in your environment")
-        return None
-
-
-if __name__ == "__main__":
-    user_id = sys.argv[1] if len(sys.argv) > 1 else "local-dev"
-    generate_token(user_id)

inspect_mcp_tools.py

@@ -1,24 +0,0 @@
-
-from backend.src.mcp.server import mcp
-import inspect
-
-def inspect_fastmcp():
-    print("MCP Object:", mcp)
-    print("Dir MCP:", dir(mcp))
-    
-    # Try to find tools
-    if hasattr(mcp, 'tools'):
-        print("Tools attr:", mcp.tools)
-    
-    if hasattr(mcp, '_tools'):
-        print("_tools attr:", mcp._tools)
-        
-    if hasattr(mcp, '_mcp_server'):
-        print("_mcp_server:", mcp._mcp_server)
-        print("Dir _mcp_server:", dir(mcp._mcp_server))
-
-if __name__ == "__main__":
-    try:
-        inspect_fastmcp()
-    except Exception as e:
-        print(f"Error: {e}")

inspect_mcp_types.py

@@ -1,9 +0,0 @@
-
-from mcp.types import CallToolResult
-import inspect
-
-def inspect_call_tool_result():
-    print("Fields:", CallToolResult.model_fields.keys())
-
-if __name__ == "__main__":
-    inspect_call_tool_result()

inspect_tool_manager.py

@@ -1,23 +0,0 @@
-
-from backend.src.mcp.server import mcp
-import inspect
-
-def inspect_tool_manager():
-    tm = mcp._tool_manager
-    print("Tool Manager:", tm)
-    print("Dir TM:", dir(tm))
-    
-    if hasattr(tm, '_tools'):
-        print("Tools dict keys:", tm._tools.keys())
-        # Inspect one tool
-        if 'read_note' in tm._tools:
-            tool = tm._tools['read_note']
-            print("Tool object:", tool)
-            print("Dir Tool:", dir(tool))
-            # Check if it has a description or attributes we can patch
-
-if __name__ == "__main__":
-    try:
-        inspect_tool_manager()
-    except Exception as e:
-        print(f"Error: {e}")

test_remote_mcp.py

@@ -1,54 +0,0 @@
-
-import requests
-import json
-import sys
-
-# Use the public HF Space URL
-BASE_URL = "https://bigwolfe-document-mcp.hf.space/mcp"
-
-def test_search_notes():
-    print("--- Testing tools/call search_notes ---")
-    payload = {
-        "jsonrpc": "2.0",
-        "method": "tools/call",
-        "params": {
-            "name": "search_notes",
-            "arguments": {"query": "API"}
-        },
-        "id": 1
-    }
-    headers = {
-        "Content-Type": "application/json",
-        "Accept": "application/json, text/event-stream"
-    }
-    
-    try:
-        response = requests.post(BASE_URL, json=payload, headers=headers, timeout=10)
-        print(f"Status: {response.status_code}")
-        try:
-            data = response.json()
-            # print(json.dumps(data, indent=2))
-            
-            res = data.get("result", {})
-            if "structuredContent" in res:
-                print("✅ structuredContent found in result")
-                print("Keys:", res["structuredContent"].keys())
-            else:
-                print("❌ structuredContent NOT found in result")
-                print(json.dumps(res, indent=2))
-                
-            if "_meta" in res:
-                print("✅ _meta found in result")
-            else:
-                print("❌ _meta NOT found in result")
-
-        except json.JSONDecodeError:
-            print("Response not JSON:", response.text)
-            
-    except Exception as e:
-        print(f"Request failed: {e}")
-
-if __name__ == "__main__":
-    # test_read_note()
-    test_search_notes()
-    # test_read_resource()