Google Chat

keyring>=24.0.0

#!/usr/bin/env python3
"""
OAuth token management for Google Chat API.
Standalone authentication - does not require the MCP server.
Cross-platform secure storage using keyring library.
"""

import base64
import http.server
import json
import secrets
import socket
import sys
import time
import webbrowser
from dataclasses import dataclass
from typing import Optional
from urllib.parse import urlencode, parse_qs, urlparse
import urllib.request
import urllib.error

import keyring

# OAuth Configuration - uses same cloud function as MCP server
CLIENT_ID = "338689075775-o75k922vn5fdl18qergr96rp8g63e4d7.apps.googleusercontent.com"
CLOUD_FUNCTION_URL = "https://google-workspace-extension.geminicli.com"
REFRESH_ENDPOINT = f"{CLOUD_FUNCTION_URL}/refreshToken"

# Keyring configuration
KEYCHAIN_SERVICE = "google-chat-skill-oauth"
KEYCHAIN_ACCOUNT = "main-account"

# Google Chat requires these scopes
SCOPES = [
    "https://www.googleapis.com/auth/chat.spaces",
    "https://www.googleapis.com/auth/chat.messages",
    "https://www.googleapis.com/auth/chat.memberships",
    "https://www.googleapis.com/auth/userinfo.profile",
]

TOKEN_EXPIRY_BUFFER_MS = 5 * 60 * 1000  # 5 minutes


# =============================================================================
# Token data classes
# =============================================================================


@dataclass
class TokenInfo:
    access_token: str
    refresh_token: Optional[str]
    expires_at: Optional[int]
    scope: Optional[str]


def get_available_port() -> int:
    """Find an available port for the OAuth callback server."""
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
        s.bind(('localhost', 0))
        return s.getsockname()[1]


# =============================================================================
# Cross-platform token storage using keyring (public API)
# =============================================================================


def get_tokens_from_keychain() -> Optional[TokenInfo]:
    """Retrieve OAuth tokens from secure storage using keyring."""
    try:
        data_str = keyring.get_password(KEYCHAIN_SERVICE, KEYCHAIN_ACCOUNT)
        if not data_str:
            return None
        data = json.loads(data_str)
        token = data.get("token", {})
        return TokenInfo(
            access_token=token.get("accessToken", ""),
            refresh_token=token.get("refreshToken"),
            expires_at=token.get("expiresAt"),
            scope=token.get("scope")
        )
    except (json.JSONDecodeError, keyring.errors.KeyringError) as e:
        print(f"Error reading tokens: {e}", file=sys.stderr)
        return None


def save_tokens_to_keychain(token_info: TokenInfo) -> bool:
    """Save OAuth tokens to secure storage using keyring."""
    data = {
        "serverName": KEYCHAIN_ACCOUNT,
        "token": {
            "accessToken": token_info.access_token,
            "refreshToken": token_info.refresh_token,
            "tokenType": "Bearer",
            "scope": token_info.scope,
            "expiresAt": token_info.expires_at
        },
        "updatedAt": int(time.time() * 1000)
    }
    try:
        keyring.set_password(KEYCHAIN_SERVICE, KEYCHAIN_ACCOUNT, json.dumps(data))
        return True
    except keyring.errors.KeyringError as e:
        print(f"Error saving tokens: {e}", file=sys.stderr)
        return False


def clear_tokens_from_keychain() -> bool:
    """Clear OAuth tokens from secure storage."""
    try:
        keyring.delete_password(KEYCHAIN_SERVICE, KEYCHAIN_ACCOUNT)
        return True
    except keyring.errors.PasswordDeleteError:
        return False
    except keyring.errors.KeyringError as e:
        print(f"Error clearing tokens: {e}", file=sys.stderr)
        return False


def is_token_expired(token_info: TokenInfo) -> bool:
    """Check if the access token is expired or expiring soon."""
    if not token_info.expires_at:
        return False
    current_time_ms = int(time.time() * 1000)
    return token_info.expires_at < (current_time_ms + TOKEN_EXPIRY_BUFFER_MS)


def refresh_access_token(refresh_token: str) -> Optional[dict]:
    """Refresh access token using the cloud function."""
    try:
        data = json.dumps({"refresh_token": refresh_token}).encode('utf-8')
        req = urllib.request.Request(
            REFRESH_ENDPOINT,
            data=data,
            headers={"Content-Type": "application/json"},
            method="POST"
        )

        with urllib.request.urlopen(req, timeout=30) as response:
            return json.loads(response.read().decode('utf-8'))
    except (urllib.error.URLError, urllib.error.HTTPError, json.JSONDecodeError) as e:
        print(f"Error refreshing token: {e}", file=sys.stderr)
        return None


class OAuthCallbackHandler(http.server.BaseHTTPRequestHandler):
    """HTTP request handler for OAuth callback."""

    token_info: Optional[TokenInfo] = None
    csrf_token: str = ""
    error: Optional[str] = None

    def log_message(self, format: str, *args) -> None:  # noqa: A002
        """Suppress default logging."""
        _ = format, args  # Unused

    def do_GET(self):
        """Handle the OAuth callback."""
        parsed = urlparse(self.path)

        if not parsed.path.startswith("/oauth2callback"):
            self.send_error(404)
            return

        params = parse_qs(parsed.query)

        # Validate CSRF token
        state = params.get("state", [""])[0]
        if state != OAuthCallbackHandler.csrf_token:
            OAuthCallbackHandler.error = "State mismatch - possible CSRF attack"
            self.send_response(400)
            self.send_header("Content-type", "text/html")
            self.end_headers()
            self.wfile.write(b"<html><body><h1>Error: State mismatch</h1></body></html>")
            return

        # Check for errors
        if "error" in params:
            OAuthCallbackHandler.error = params.get("error_description", params["error"])[0]
            self.send_response(400)
            self.send_header("Content-type", "text/html")
            self.end_headers()
            self.wfile.write(f"<html><body><h1>Error: {OAuthCallbackHandler.error}</h1></body></html>".encode())
            return

        # Extract tokens
        access_token = params.get("access_token", [""])[0]
        refresh_token = params.get("refresh_token", [""])[0]
        expiry_date = params.get("expiry_date", [""])[0]
        scope = params.get("scope", [""])[0]

        if access_token:
            OAuthCallbackHandler.token_info = TokenInfo(
                access_token=access_token,
                refresh_token=refresh_token or None,
                expires_at=int(expiry_date) if expiry_date else None,
                scope=scope or None
            )

            self.send_response(200)
            self.send_header("Content-type", "text/html")
            self.end_headers()
            self.wfile.write(b"""<html><body>
                <h1>Authentication successful!</h1>
                <p>You can close this window and return to the terminal.</p>
                <script>window.close();</script>
            </body></html>""")
        else:
            OAuthCallbackHandler.error = "No access token received"
            self.send_response(400)
            self.send_header("Content-type", "text/html")
            self.end_headers()
            self.wfile.write(b"<html><body><h1>Error: No access token received</h1></body></html>")


def perform_oauth_flow() -> Optional[TokenInfo]:
    """Perform the full OAuth flow with browser-based authentication."""
    port = get_available_port()
    redirect_uri = f"http://localhost:{port}/oauth2callback"
    csrf_token = secrets.token_hex(32)

    # Build state payload for cloud function
    state_payload = {
        "uri": redirect_uri,
        "manual": False,
        "csrf": csrf_token
    }
    state = base64.b64encode(json.dumps(state_payload).encode()).decode()

    # Build auth URL
    auth_params = {
        "client_id": CLIENT_ID,
        "redirect_uri": CLOUD_FUNCTION_URL,  # Cloud function handles the secret
        "response_type": "code",
        "scope": " ".join(SCOPES),
        "access_type": "offline",
        "prompt": "consent",
        "state": state
    }
    auth_url = f"https://accounts.google.com/o/oauth2/v2/auth?{urlencode(auth_params)}"

    # Set up callback handler
    OAuthCallbackHandler.csrf_token = csrf_token
    OAuthCallbackHandler.token_info = None
    OAuthCallbackHandler.error = None

    # Start local server
    server = http.server.HTTPServer(('localhost', port), OAuthCallbackHandler)
    server.timeout = 300  # 5 minute timeout

    print(f"Opening browser for authentication...", file=sys.stderr)
    print(f"If browser doesn't open, visit: {auth_url}", file=sys.stderr)

    webbrowser.open(auth_url)

    # Wait for callback
    print("Waiting for authentication...", file=sys.stderr)
    while OAuthCallbackHandler.token_info is None and OAuthCallbackHandler.error is None:
        server.handle_request()

    server.server_close()

    if OAuthCallbackHandler.error:
        print(f"Authentication failed: {OAuthCallbackHandler.error}", file=sys.stderr)
        return None

    return OAuthCallbackHandler.token_info


def get_valid_access_token(interactive: bool = True) -> Optional[str]:
    """
    Get a valid access token.

    Args:
        interactive: If True, will prompt for OAuth flow if no tokens exist.
                    If False, will return None if no valid tokens.
    """
    token_info = get_tokens_from_keychain()

    # No tokens - need to authenticate
    if not token_info or not token_info.access_token:
        if not interactive:
            return None

        print("No OAuth tokens found. Starting authentication...", file=sys.stderr)
        token_info = perform_oauth_flow()

        if not token_info:
            return None

        save_tokens_to_keychain(token_info)
        print("Authentication successful!", file=sys.stderr)
        return token_info.access_token

    # Check if token needs refresh
    if is_token_expired(token_info) and token_info.refresh_token:
        print("Access token expired, refreshing...", file=sys.stderr)
        new_tokens = refresh_access_token(token_info.refresh_token)

        if new_tokens:
            token_info.access_token = new_tokens.get("access_token", token_info.access_token)
            token_info.expires_at = new_tokens.get("expiry_date", token_info.expires_at)
            save_tokens_to_keychain(token_info)
            print("Token refreshed successfully.", file=sys.stderr)
        else:
            # Refresh failed - try re-authenticating if interactive
            if interactive:
                print("Token refresh failed. Re-authenticating...", file=sys.stderr)
                token_info = perform_oauth_flow()
                if token_info:
                    save_tokens_to_keychain(token_info)
                else:
                    return None
            else:
                print("Warning: Failed to refresh token.", file=sys.stderr)

    return token_info.access_token if token_info else None


def logout() -> bool:
    """Clear stored OAuth tokens."""
    return clear_tokens_from_keychain()


def main():
    """CLI for auth operations."""
    import argparse

    parser = argparse.ArgumentParser(description="Google Chat OAuth management")
    subparsers = parser.add_subparsers(dest="command")

    subparsers.add_parser("login", help="Authenticate with Google")
    subparsers.add_parser("logout", help="Clear stored tokens")
    subparsers.add_parser("token", help="Print current access token")
    subparsers.add_parser("status", help="Check authentication status")

    args = parser.parse_args()

    if args.command == "login":
        token_info = perform_oauth_flow()
        if token_info:
            save_tokens_to_keychain(token_info)
            print("Login successful!")
        else:
            print("Login failed.")
            sys.exit(1)

    elif args.command == "logout":
        if logout():
            print("Logged out successfully.")
        else:
            print("No tokens to clear.")

    elif args.command == "token":
        token = get_valid_access_token(interactive=False)
        if token:
            print(token)
        else:
            print("Not authenticated. Run: python auth.py login", file=sys.stderr)
            sys.exit(1)

    elif args.command == "status":
        token_info = get_tokens_from_keychain()
        if token_info and token_info.access_token:
            expired = is_token_expired(token_info)
            print(f"Status: Authenticated")
            print(f"Token expired: {expired}")
            if token_info.expires_at:
                expires_in = (token_info.expires_at - int(time.time() * 1000)) / 1000 / 60
                print(f"Expires in: {expires_in:.1f} minutes")
        else:
            print("Status: Not authenticated")
            print("Run: python auth.py login")
            sys.exit(1)

    else:
        parser.print_help()


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Google Chat API operations.
Lightweight alternative to the full Google Workspace MCP server.
"""

import argparse
import json
import mimetypes
import os
import sys
import urllib.request
import urllib.error
import urllib.parse
from typing import Optional

from auth import get_valid_access_token

CHAT_API_BASE = "https://chat.googleapis.com/v1"
CHAT_UPLOAD_BASE = "https://chat.googleapis.com/upload/v1"


def api_request(method: str, endpoint: str, data: Optional[dict] = None, params: Optional[dict] = None) -> dict:
    """Make an authenticated request to the Google Chat API."""
    token = get_valid_access_token()
    if not token:
        return {"error": "Failed to get access token"}

    url = f"{CHAT_API_BASE}/{endpoint}"
    if params:
        url += "?" + urllib.parse.urlencode(params)

    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json"
    }

    body = json.dumps(data).encode('utf-8') if data else None

    try:
        req = urllib.request.Request(url, data=body, headers=headers, method=method)
        with urllib.request.urlopen(req, timeout=30) as response:
            return json.loads(response.read().decode('utf-8'))
    except urllib.error.HTTPError as e:
        error_body = e.read().decode('utf-8') if e.fp else str(e)
        return {"error": f"HTTP {e.code}: {error_body}"}
    except urllib.error.URLError as e:
        return {"error": f"Request failed: {e.reason}"}
    except json.JSONDecodeError:
        return {"error": "Invalid JSON response"}


def upload_attachment(space_name: str, file_path: str, text: str = "") -> dict:
    """Send a message with a file attachment (two-step: upload then send)."""
    import requests as req_lib

    token = get_valid_access_token()
    if not token:
        return {"error": "Failed to get access token"}

    if not os.path.isfile(file_path):
        return {"error": f"File not found: {file_path}"}

    mime_type, _ = mimetypes.guess_type(file_path)
    if not mime_type:
        mime_type = "application/octet-stream"

    filename = os.path.basename(file_path)
    headers = {"Authorization": f"Bearer {token}"}

    # Step 1: Upload the file to get an attachment token
    upload_url = f"{CHAT_UPLOAD_BASE}/{space_name}/attachments:upload"
    metadata = json.dumps({"filename": filename})

    with open(file_path, 'rb') as f:
        upload_resp = req_lib.post(
            upload_url,
            headers=headers,
            files={
                "metadata": ("metadata", metadata, "application/json"),
                "file": (filename, f, mime_type),
            },
            params={"uploadType": "multipart"},
            timeout=60,
        )

    if upload_resp.status_code != 200:
        return {"error": f"Upload failed HTTP {upload_resp.status_code}: {upload_resp.text}"}

    upload_data = upload_resp.json()
    attachment_token = upload_data.get("attachmentDataRef", {}).get("attachmentUploadToken")
    if not attachment_token:
        return {"error": "Upload succeeded but no attachment token returned"}

    # Step 2: Send message with the attachment reference
    msg_url = f"{CHAT_API_BASE}/{space_name}/messages"
    msg_data = {
        "text": text,
        "attachment": [{
            "contentName": filename,
            "contentType": mime_type,
            "attachmentDataRef": {
                "attachmentUploadToken": attachment_token
            }
        }]
    }

    msg_resp = req_lib.post(
        msg_url,
        headers={**headers, "Content-Type": "application/json"},
        data=json.dumps(msg_data),
        timeout=30,
    )

    if msg_resp.status_code != 200:
        return {"error": f"Send failed HTTP {msg_resp.status_code}: {msg_resp.text}"}

    return msg_resp.json()


def list_spaces() -> dict:
    """List all spaces the user is a member of."""
    result = api_request("GET", "spaces")
    return result.get("spaces", []) if "spaces" in result else result


def find_space_by_name(display_name: str) -> dict:
    """Find a space by its display name."""
    spaces = list_spaces()
    if isinstance(spaces, dict) and "error" in spaces:
        return spaces

    matching = [s for s in spaces if s.get("displayName") == display_name]
    if matching:
        return {"spaces": matching}
    return {"error": f"No space found with display name: {display_name}"}


def get_messages(space_name: str, page_size: int = 25, page_token: Optional[str] = None,
                 order_by: str = "createTime desc") -> dict:
    """Get messages from a space."""
    params = {"pageSize": page_size, "orderBy": order_by}
    if page_token:
        params["pageToken"] = page_token

    return api_request("GET", f"{space_name}/messages", params=params)


def send_message(space_name: str, text: str, attachment: Optional[str] = None) -> dict:
    """Send a message to a space, optionally with a file attachment."""
    if attachment:
        return upload_attachment(space_name, attachment, text)
    return api_request("POST", f"{space_name}/messages", data={"text": text})


def send_dm(email: str, text: str, attachment: Optional[str] = None) -> dict:
    """Send a direct message to a user by email, optionally with a file attachment."""
    # First, set up or find the DM space
    space_data = {
        "space": {"spaceType": "DIRECT_MESSAGE"},
        "memberships": [{"member": {"name": f"users/{email}", "type": "HUMAN"}}]
    }
    space_result = api_request("POST", "spaces:setup", data=space_data)

    if "error" in space_result:
        return space_result

    space_name = space_result.get("name")
    if not space_name:
        return {"error": "Failed to create DM space"}

    # Send the message (with or without attachment)
    if attachment:
        return upload_attachment(space_name, attachment, text)
    return api_request("POST", f"{space_name}/messages", data={"text": text})


def find_dm_by_email(email: str) -> dict:
    """Find or create a DM space with a user."""
    space_data = {
        "space": {"spaceType": "DIRECT_MESSAGE"},
        "memberships": [{"member": {"name": f"users/{email}", "type": "HUMAN"}}]
    }
    return api_request("POST", "spaces:setup", data=space_data)


def list_threads(space_name: str, page_size: int = 25, page_token: Optional[str] = None) -> dict:
    """List threads from a space."""
    params = {"pageSize": page_size, "orderBy": "createTime desc"}
    if page_token:
        params["pageToken"] = page_token

    result = api_request("GET", f"{space_name}/messages", params=params)

    if "error" in result:
        return result

    # Group messages by thread
    messages = result.get("messages", [])
    seen_threads = set()
    threads = []

    for msg in messages:
        thread_name = msg.get("thread", {}).get("name")
        if thread_name and thread_name not in seen_threads:
            threads.append(msg)
            seen_threads.add(thread_name)

    return {"threads": threads, "nextPageToken": result.get("nextPageToken")}


def setup_space(display_name: str, user_emails: list) -> dict:
    """Create a new space with members."""
    memberships = [
        {"member": {"name": f"users/{email}", "type": "HUMAN"}}
        for email in user_emails
    ]
    space_data = {
        "space": {"spaceType": "SPACE", "displayName": display_name},
        "memberships": memberships
    }
    return api_request("POST", "spaces:setup", data=space_data)


def main():
    parser = argparse.ArgumentParser(description="Google Chat API operations")
    subparsers = parser.add_subparsers(dest="command", required=True)

    # list-spaces
    subparsers.add_parser("list-spaces", help="List all spaces")

    # find-space
    find_space_parser = subparsers.add_parser("find-space", help="Find a space by display name")
    find_space_parser.add_argument("name", help="Display name of the space")

    # get-messages
    get_messages_parser = subparsers.add_parser("get-messages", help="Get messages from a space")
    get_messages_parser.add_argument("space", help="Space name (e.g., spaces/AAAA123)")
    get_messages_parser.add_argument("--limit", type=int, default=25, help="Max messages to return")
    get_messages_parser.add_argument("--page-token", help="Pagination token")

    # send-message
    send_message_parser = subparsers.add_parser("send-message", help="Send a message to a space")
    send_message_parser.add_argument("space", help="Space name (e.g., spaces/AAAA123)")
    send_message_parser.add_argument("text", nargs="?", default="", help="Message text")
    send_message_parser.add_argument("--attachment", help="Path to file to attach")

    # send-dm
    send_dm_parser = subparsers.add_parser("send-dm", help="Send a direct message")
    send_dm_parser.add_argument("email", help="Recipient email address")
    send_dm_parser.add_argument("text", nargs="?", default="", help="Message text")
    send_dm_parser.add_argument("--attachment", help="Path to file to attach")

    # find-dm
    find_dm_parser = subparsers.add_parser("find-dm", help="Find or create DM space")
    find_dm_parser.add_argument("email", help="User's email address")

    # list-threads
    list_threads_parser = subparsers.add_parser("list-threads", help="List threads in a space")
    list_threads_parser.add_argument("space", help="Space name")
    list_threads_parser.add_argument("--limit", type=int, default=25, help="Max threads to return")

    # setup-space
    setup_space_parser = subparsers.add_parser("setup-space", help="Create a new space")
    setup_space_parser.add_argument("name", help="Display name for the space")
    setup_space_parser.add_argument("emails", nargs="+", help="Member email addresses")

    args = parser.parse_args()

    if args.command == "list-spaces":
        result = list_spaces()
    elif args.command == "find-space":
        result = find_space_by_name(args.name)
    elif args.command == "get-messages":
        result = get_messages(args.space, args.limit, args.page_token)
    elif args.command == "send-message":
        result = send_message(args.space, args.text, getattr(args, 'attachment', None))
    elif args.command == "send-dm":
        result = send_dm(args.email, args.text, getattr(args, 'attachment', None))
    elif args.command == "find-dm":
        result = find_dm_by_email(args.email)
    elif args.command == "list-threads":
        result = list_threads(args.space, args.limit)
    elif args.command == "setup-space":
        result = setup_space(args.name, args.emails)
    else:
        result = {"error": f"Unknown command: {args.command}"}

    print(json.dumps(result, indent=2))

    if isinstance(result, dict) and "error" in result:
        sys.exit(1)


if __name__ == "__main__":
    main()