t66_langmem/scripts/verify-signal-auth.py

#!/usr/bin/env python3
"""
Verify Signal bridge authentication status and test functionality
"""

import asyncio
import httpx
import json

MATRIX_HOMESERVER = "https://matrix.klas.chat"
CLAUDE_ACCESS_TOKEN = "syt_Y2xhdWRl_CoBgPoHbtMOxhvOUcMnz_2WRPZJ"
SIGNAL_BRIDGE_BOT_ID = "@signalbot:matrix.klas.chat"
BRIDGE_DM_ROOM_ID = "!oBnnfKDprgMEHNhNjL:matrix.klas.chat"

async def send_bridge_command(command, explanation=""):
    """Send a command to the Signal bridge bot"""
    try:
        async with httpx.AsyncClient() as client:
            headers = {
                "Authorization": f"Bearer {CLAUDE_ACCESS_TOKEN}",
                "Content-Type": "application/json"
            }
            
            print(f"🤖 Sending: {command}")
            if explanation:
                print(f"   {explanation}")
            
            response = await client.post(
                f"{MATRIX_HOMESERVER}/_matrix/client/v3/rooms/{BRIDGE_DM_ROOM_ID}/send/m.room.message",
                headers=headers,
                json={
                    "msgtype": "m.text",
                    "body": command
                }
            )
            
            if response.status_code == 200:
                print(f"✅ Command sent: {command}")
                await asyncio.sleep(2)
                return True
            else:
                print(f"❌ Failed to send command: {response.status_code}")
                return False
                
    except Exception as e:
        print(f"❌ Error sending command: {e}")
        return False

async def get_bridge_room_messages(limit=10):
    """Get recent messages from the Signal bridge DM room"""
    try:
        async with httpx.AsyncClient() as client:
            headers = {"Authorization": f"Bearer {CLAUDE_ACCESS_TOKEN}"}
            
            response = await client.get(
                f"{MATRIX_HOMESERVER}/_matrix/client/v3/rooms/{BRIDGE_DM_ROOM_ID}/messages",
                headers=headers,
                params={"limit": limit, "dir": "b"}
            )
            
            if response.status_code == 200:
                data = response.json()
                return data.get("chunk", [])
            else:
                print(f"❌ Failed to get messages: {response.status_code}")
                return []
                
    except Exception as e:
        print(f"❌ Error getting bridge messages: {e}")
        return []

async def check_authentication_status():
    """Check if authentication is complete"""
    print("🔍 Checking authentication status...")
    
    # Check login status
    await send_bridge_command("list-logins", "Check current logins")
    
    messages = await get_bridge_room_messages(5)
    recent_bot_messages = [
        msg for msg in messages 
        if msg.get("sender") == SIGNAL_BRIDGE_BOT_ID 
        and msg.get("type") == "m.room.message"
    ]
    
    print("🔔 Recent bot responses:")
    authenticated = False
    for msg in recent_bot_messages[:3]:
        body = msg.get("content", {}).get("body", "")
        print(f"   → {body}")
        
        # Check for signs of successful authentication
        if "login" in body.lower() and "successful" in body.lower():
            authenticated = True
        elif "logged in" in body.lower():
            authenticated = True
        elif "sgnl://linkdevice" in body:
            print("   ⚠️  QR code still present - authentication not complete")
        elif "no logins" in body.lower():
            print("   ⚠️  No logins found - authentication not complete")
    
    return authenticated

async def test_bridge_functionality():
    """Test basic bridge functionality"""
    print("\n🧪 Testing bridge functionality...")
    
    # Test version command
    await send_bridge_command("version", "Get bridge version")
    
    # Test help command
    await send_bridge_command("help", "Get help information")
    
    messages = await get_bridge_room_messages(5)
    recent_bot_messages = [
        msg for msg in messages 
        if msg.get("sender") == SIGNAL_BRIDGE_BOT_ID 
        and msg.get("type") == "m.room.message"
    ]
    
    print("🔔 Functionality test responses:")
    for msg in recent_bot_messages[:3]:
        body = msg.get("content", {}).get("body", "")
        print(f"   → {body[:100]}...")  # Truncate long responses
    
    return len(recent_bot_messages) > 0

async def main():
    """Main function"""
    print("🔍 Signal Bridge Authentication Verification")
    print("=" * 50)
    
    # Check authentication status
    is_authenticated = await check_authentication_status()
    
    if is_authenticated:
        print("\n✅ Authentication appears to be successful!")
        
        # Test functionality
        functionality_works = await test_bridge_functionality()
        
        if functionality_works:
            print("\n✅ Bridge functionality test passed!")
            print("🚀 Signal bridge is ready for use")
        else:
            print("\n❌ Bridge functionality test failed")
    else:
        print("\n⚠️  Authentication not yet complete")
        print("📱 Please scan the QR code with your Signal app to complete authentication")
        print("🔗 Check the previous script output for the QR code link")
    
    print("\n" + "=" * 50)
    print("🎯 Verification complete")

if __name__ == "__main__":
    asyncio.run(main())