furiosa
dc5877096a
- Add rbw integration to retrieve credentials from Bitwarden vault - Automate email/password entry with Selenium WebDriver - Handle 2FA by falling back to manual completion - Add clipboard support for Wayland (wl-copy) and X11 (xclip) - Add CLI flags: --entry, --no-copy, --manual - Add DESIGN.md documenting the implementation approach
248 lines
7.6 KiB
Python
248 lines
7.6 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Google Cookie Retrieval - Automated login with rbw password lookup.
|
|
|
|
Automates Google login for Mautrix Google Chat bridge authentication.
|
|
"""
|
|
import argparse
|
|
import json
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
|
|
from selenium import webdriver
|
|
from selenium.webdriver.chrome.service import Service
|
|
from selenium.webdriver.common.by import By
|
|
from selenium.webdriver.common.keys import Keys
|
|
from selenium.webdriver.support import expected_conditions as EC
|
|
from selenium.webdriver.support.ui import WebDriverWait
|
|
|
|
|
|
def check_rbw_unlocked() -> bool:
|
|
"""Check if rbw vault is unlocked."""
|
|
result = subprocess.run(['rbw', 'unlocked'], capture_output=True)
|
|
return result.returncode == 0
|
|
|
|
|
|
def prompt_rbw_unlock() -> bool:
|
|
"""Prompt user to unlock rbw vault. Returns True if successful."""
|
|
print("rbw vault is locked. Unlocking...")
|
|
result = subprocess.run(['rbw', 'unlock'])
|
|
return result.returncode == 0
|
|
|
|
|
|
def get_google_credentials(entry_name: str) -> tuple[str, str]:
|
|
"""Get username and password from rbw."""
|
|
try:
|
|
username = subprocess.run(
|
|
['rbw', 'get', '-f', 'username', entry_name],
|
|
capture_output=True, text=True, check=True
|
|
).stdout.strip()
|
|
|
|
password = subprocess.run(
|
|
['rbw', 'get', entry_name],
|
|
capture_output=True, text=True, check=True
|
|
).stdout.strip()
|
|
|
|
return username, password
|
|
except subprocess.CalledProcessError as e:
|
|
print(f"Error retrieving credentials for '{entry_name}': {e.stderr}")
|
|
sys.exit(1)
|
|
|
|
|
|
def copy_to_clipboard(text: str) -> bool:
|
|
"""Copy text to clipboard. Tries wl-copy (Wayland) first, falls back to xclip (X11)."""
|
|
# Try wl-copy first (Wayland)
|
|
try:
|
|
result = subprocess.run(['wl-copy'], input=text.encode(), check=True)
|
|
return True
|
|
except FileNotFoundError:
|
|
pass
|
|
except subprocess.CalledProcessError:
|
|
pass
|
|
|
|
# Fall back to xclip (X11)
|
|
try:
|
|
process = subprocess.Popen(
|
|
['xclip', '-selection', 'clipboard'],
|
|
stdin=subprocess.PIPE
|
|
)
|
|
process.communicate(text.encode())
|
|
return process.returncode == 0
|
|
except FileNotFoundError:
|
|
print("Warning: Neither wl-copy nor xclip found, cannot copy to clipboard")
|
|
return False
|
|
|
|
|
|
def wait_for_chat_page(driver, timeout: int = 120) -> bool:
|
|
"""Wait for successful login by checking for chat.google.com content."""
|
|
try:
|
|
WebDriverWait(driver, timeout).until(
|
|
lambda d: "chat.google.com" in d.current_url and
|
|
"accounts.google.com" not in d.current_url
|
|
)
|
|
return True
|
|
except Exception:
|
|
return False
|
|
|
|
|
|
def automate_login(driver, username: str, password: str) -> bool:
|
|
"""
|
|
Automate Google login flow.
|
|
Returns True if login succeeded, False if 2FA or other intervention needed.
|
|
"""
|
|
print(f"Logging in as {username}...")
|
|
|
|
# Navigate to Google sign-in
|
|
driver.get("https://accounts.google.com/signin/v2/identifier?continue=https://chat.google.com")
|
|
|
|
try:
|
|
# Wait for and enter email
|
|
email_input = WebDriverWait(driver, 10).until(
|
|
EC.presence_of_element_located((By.ID, "identifierId"))
|
|
)
|
|
email_input.send_keys(username)
|
|
email_input.send_keys(Keys.RETURN)
|
|
|
|
# Wait for password page
|
|
password_input = WebDriverWait(driver, 10).until(
|
|
EC.presence_of_element_located((By.NAME, "Passwd"))
|
|
)
|
|
time.sleep(0.5) # Brief pause for page transition
|
|
password_input.send_keys(password)
|
|
password_input.send_keys(Keys.RETURN)
|
|
|
|
# Check if we landed on chat or got stuck (2FA, etc.)
|
|
time.sleep(2)
|
|
|
|
# If we're still on accounts.google.com, 2FA or challenge is required
|
|
if "accounts.google.com" in driver.current_url:
|
|
print("\n2FA or additional verification required.")
|
|
print("Please complete the verification in the browser.")
|
|
return False
|
|
|
|
return True
|
|
|
|
except Exception as e:
|
|
print(f"Login automation error: {e}")
|
|
return False
|
|
|
|
|
|
def extract_cookies(driver) -> dict:
|
|
"""Extract the required cookies from the browser."""
|
|
# Navigate to an invalid URL to avoid redirects
|
|
driver.get("https://chat.google.com/u/0/mole/world")
|
|
time.sleep(1)
|
|
|
|
all_cookies = driver.get_cookies()
|
|
target_names = {"COMPASS", "SSID", "SID", "OSID", "HSID"}
|
|
extracted = {}
|
|
compass_cookie = None
|
|
|
|
for cookie in all_cookies:
|
|
name_upper = cookie["name"].upper()
|
|
if name_upper not in target_names:
|
|
continue
|
|
if name_upper == "COMPASS":
|
|
if cookie.get("path", "") == "/":
|
|
compass_cookie = cookie
|
|
elif compass_cookie is None:
|
|
compass_cookie = cookie
|
|
else:
|
|
extracted[name_upper] = cookie["value"]
|
|
|
|
if compass_cookie:
|
|
extracted["COMPASS"] = compass_cookie["value"]
|
|
|
|
return extracted
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(
|
|
description="Google Cookie Retrieval - Automated login with rbw"
|
|
)
|
|
parser.add_argument(
|
|
'--entry', '-e',
|
|
default='google.com',
|
|
help='rbw entry name for Google credentials (default: google.com)'
|
|
)
|
|
parser.add_argument(
|
|
'--no-copy',
|
|
action='store_true',
|
|
help='Do not copy cookies to clipboard'
|
|
)
|
|
parser.add_argument(
|
|
'--manual', '-m',
|
|
action='store_true',
|
|
help='Skip auto-login, use manual flow'
|
|
)
|
|
args = parser.parse_args()
|
|
|
|
# Check and unlock rbw if needed
|
|
if not args.manual:
|
|
if not check_rbw_unlocked():
|
|
if not prompt_rbw_unlock():
|
|
print("Failed to unlock rbw vault.")
|
|
sys.exit(1)
|
|
|
|
username, password = get_google_credentials(args.entry)
|
|
|
|
# Configure Chrome options
|
|
options = webdriver.ChromeOptions()
|
|
options.add_argument("--incognito")
|
|
|
|
# Initialize Chrome driver
|
|
service = Service()
|
|
driver = webdriver.Chrome(service=service, options=options)
|
|
|
|
try:
|
|
if args.manual:
|
|
# Manual flow (original behavior)
|
|
print("Opening https://chat.google.com ...")
|
|
driver.get("https://chat.google.com")
|
|
print("\nA new incognito window has been opened.")
|
|
print("Please log in normally.")
|
|
print("\nOnce logged in, press Enter to extract cookies...")
|
|
input()
|
|
else:
|
|
# Automated flow
|
|
success = automate_login(driver, username, password)
|
|
|
|
if not success:
|
|
# Wait for manual 2FA completion
|
|
print("\nWaiting for login to complete...")
|
|
print("Press Enter once you've completed verification...")
|
|
input()
|
|
|
|
# Verify we're logged in
|
|
if not wait_for_chat_page(driver, timeout=10):
|
|
print("Warning: May not be fully logged in. Attempting cookie extraction anyway.")
|
|
|
|
# Extract cookies
|
|
cookies = extract_cookies(driver)
|
|
|
|
if not cookies:
|
|
print("Error: No cookies extracted. Login may have failed.")
|
|
sys.exit(1)
|
|
|
|
json_data = json.dumps(cookies, indent=2)
|
|
|
|
print("\nExtracted Cookie JSON:")
|
|
print(json_data)
|
|
|
|
# Copy to clipboard
|
|
if not args.no_copy:
|
|
if copy_to_clipboard(json_data):
|
|
print("\nCookies copied to clipboard!")
|
|
else:
|
|
print("\nNote: Could not copy to clipboard.")
|
|
|
|
print("\nClosing browser to preserve cookie validity...")
|
|
|
|
finally:
|
|
driver.quit()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|