the-nexus/tests/test_websocket_security.py

#!/usr/bin/env python3
"""
Test WebSocket gateway security configuration.
Issue #1514: [Security] WebSocket gateway listens on 0.0.0.0
"""

import os
import sys
import subprocess

def test_server_binding():
    """Test that server binds to correct address."""
    print("Testing server binding configuration...")
    
    # Check server.py for HOST configuration
    server_path = os.path.join(os.path.dirname(__file__), '..', 'server.py')
    
    if not os.path.exists(server_path):
        print(f"❌ ERROR: server.py not found at {server_path}")
        return False
    
    with open(server_path, 'r') as f:
        content = f.read()
    
    # Check for HOST configuration
    if 'HOST = os.environ.get("NEXUS_WS_HOST", "127.0.0.1")' in content:
        print("✅ HOST configured to use environment variable with 127.0.0.1 default")
    else:
        print("❌ HOST not properly configured")
        return False
    
    # Check for authentication code
    if 'Authentication check for external connections' in content:
        print("✅ Authentication check implemented")
    else:
        print("❌ Authentication check not found")
        return False
    
    # Check for token validation
    if 'auth_data.get("token")' in content:
        print("✅ Token validation implemented")
    else:
        print("❌ Token validation not found")
        return False
    
    return True

def test_environment_variables():
    """Test environment variable configuration."""
    print("\nTesting environment variable configuration...")
    
    # Check server.py for environment variable usage
    server_path = os.path.join(os.path.dirname(__file__), '..', 'server.py')
    
    if not os.path.exists(server_path):
        print(f"❌ ERROR: server.py not found at {server_path}")
        return False
    
    with open(server_path, 'r') as f:
        content = f.read()
    
    # Check for environment variable usage
    if 'os.environ.get("NEXUS_WS_HOST"' in content:
        print("✅ HOST uses environment variable")
    else:
        print("❌ HOST does not use environment variable")
        return False
    
    if '"127.0.0.1"' in content:
        print("✅ Default HOST is 127.0.0.1")
    else:
        print("❌ Default HOST is not 127.0.0.1")
        return False
    
    if 'os.environ.get("NEXUS_WS_PORT"' in content:
        print("✅ PORT uses environment variable")
    else:
        print("❌ PORT does not use environment variable")
        return False
    
    if 'os.environ.get("NEXUS_WS_AUTH_TOKEN"' in content:
        print("✅ Auth token uses environment variable")
    else:
        print("❌ Auth token does not use environment variable")
        return False
    
    return True

def test_client_authentication():
    """Test client authentication code."""
    print("\nTesting client authentication code...")
    
    # Check app.js for authentication code
    app_path = os.path.join(os.path.dirname(__file__), '..', 'app.js')
    
    if not os.path.exists(app_path):
        print(f"❌ ERROR: app.js not found at {app_path}")
        return False
    
    with open(app_path, 'r') as f:
        content = f.read()
    
    # Check for authentication code
    if 'isLocalhost' in content:
        print("✅ Client checks for localhost")
    else:
        print("❌ Client does not check for localhost")
        return False
    
    if 'nexus-ws-auth-token' in content:
        print("✅ Client checks for auth token in localStorage")
    else:
        print("❌ Client does not check for auth token")
        return False
    
    if 'type: \'auth\'' in content:
        print("✅ Client sends auth message")
    else:
        print("❌ Client does not send auth message")
        return False
    
    return True

def main():
    """Run all tests."""
    print("=" * 60)
    print("WebSocket Gateway Security Tests")
    print("=" * 60)
    
    tests = [
        ("Server binding configuration", test_server_binding),
        ("Environment variable configuration", test_environment_variables),
        ("Client authentication code", test_client_authentication),
    ]
    
    results = []
    for name, test_func in tests:
        print(f"\n{name}:")
        try:
            result = test_func()
            results.append((name, result))
        except Exception as e:
            print(f"❌ Test failed with exception: {e}")
            results.append((name, False))
    
    print("\n" + "=" * 60)
    print("Test Results:")
    print("=" * 60)
    
    passed = 0
    for name, result in results:
        status = "✅ PASS" if result else "❌ FAIL"
        print(f"{status}: {name}")
        if result:
            passed += 1
    
    print(f"\nPassed: {passed}/{len(results)}")
    
    if passed == len(results):
        print("\n✅ All tests passed!")
        return 0
    else:
        print("\n❌ Some tests failed!")
        return 1

if __name__ == "__main__":
    sys.exit(main())