193 lines
8.0 KiB
Python
193 lines
8.0 KiB
Python
import json
|
|
import logging
|
|
import os
|
|
import sqlite3
|
|
from typing import Dict
|
|
|
|
from playwright.sync_api import sync_playwright
|
|
|
|
logger = logging.getLogger("desktopenv.getters.chrome")
|
|
|
|
"""
|
|
WARNING:
|
|
1. Functions from this script assume that no account is registered on Chrome, otherwise the default file path needs to be changed.
|
|
2. The functions are not tested on Windows and Mac, but they should work.
|
|
"""
|
|
|
|
|
|
# The following ones just need to load info from the files of software, no need to connect to the software
|
|
def get_default_search_engine(env, config: Dict[str, str]):
|
|
os_type = env.vm_platform
|
|
if os_type == 'Windows':
|
|
preference_file_path = env.controller.execute_python_command("""import os; print(os.path.join(os.getenv('LOCALAPPDATA'),
|
|
'Google\\Chrome\\User Data\\Default\\Preferences'))""")['output'].strip()
|
|
elif os_type == 'Darwin':
|
|
preference_file_path = env.controller.execute_python_command(
|
|
"import os; print(os.path.join(os.getenv('HOME'), 'Library/Application Support/Google/Chrome/Default/Preferences'))")[
|
|
'output'].strip()
|
|
elif os_type == 'Linux':
|
|
preference_file_path = env.controller.execute_python_command(
|
|
"import os; print(os.path.join(os.getenv('HOME'), '.config/google-chrome/Default/Preferences'))")[
|
|
'output'].strip()
|
|
else:
|
|
raise Exception('Unsupported operating system')
|
|
|
|
try:
|
|
content = env.controller.get_file(preference_file_path)
|
|
data = json.loads(content)
|
|
|
|
# The path within the JSON data to the default search engine might vary
|
|
search_engine = data.get('default_search_provider_data', {}).get('template_url_data', {}).get('short_name',
|
|
'Google')
|
|
return search_engine
|
|
except Exception as e:
|
|
logger.error(f"Error: {e}")
|
|
return "Google"
|
|
|
|
|
|
def get_cookie_data(env, config: Dict[str, str]):
|
|
os_type = env.vm_platform
|
|
if os_type == 'Windows':
|
|
chrome_cookie_file_path = env.controller.execute_python_command("""import os; print(os.path.join(os.getenv('LOCALAPPDATA'),
|
|
'Google\\Chrome\\User Data\\Default\\Cookies'))""")['output'].strip()
|
|
elif os_type == 'Darwin':
|
|
chrome_cookie_file_path = env.controller.execute_python_command(
|
|
"import os; print(os.path.join(os.getenv('HOME'), 'Library/Application Support/Google/Chrome/Default/Cookies'))")[
|
|
'output'].strip()
|
|
elif os_type == 'Linux':
|
|
chrome_cookie_file_path = env.controller.execute_python_command(
|
|
"import os; print(os.path.join(os.getenv('HOME'), '.config/google-chrome/Default/Cookies'))")[
|
|
'output'].strip()
|
|
else:
|
|
raise Exception('Unsupported operating system')
|
|
|
|
# todo: add a new controller function to connect the cookie database
|
|
#############
|
|
try:
|
|
conn = sqlite3.connect(chrome_cookie_file_path)
|
|
cursor = conn.cursor()
|
|
|
|
# Query to check for OpenAI cookies
|
|
cursor.execute("SELECT * FROM cookies")
|
|
cookies = cursor.fetchall()
|
|
|
|
return cookies
|
|
except Exception as e:
|
|
logger.error(f"Error: {e}")
|
|
return None
|
|
#############
|
|
|
|
|
|
def get_bookmarks(env, config: Dict[str, str]):
|
|
os_type = env.vm_platform
|
|
if os_type == 'Windows':
|
|
preference_file_path = env.controller.execute_python_command("""import os; print(os.path.join(os.getenv('LOCALAPPDATA'),
|
|
'Google\\Chrome\\User Data\\Default\\Bookmarks'))""")['output'].strip()
|
|
elif os_type == 'Darwin':
|
|
preference_file_path = env.controller.execute_python_command(
|
|
"import os; print(os.path.join(os.getenv('HOME'), 'Library/Application Support/Google/Chrome/Default/Bookmarks'))")[
|
|
'output'].strip()
|
|
elif os_type == 'Linux':
|
|
preference_file_path = env.controller.execute_python_command(
|
|
"import os; print(os.path.join(os.getenv('HOME'), '.config/google-chrome/Default/Bookmarks'))")[
|
|
'output'].strip()
|
|
else:
|
|
raise Exception('Unsupported operating system')
|
|
|
|
try:
|
|
content = env.controller.get_file(preference_file_path)
|
|
# make content json variable
|
|
data = json.load(content)
|
|
|
|
bookmarks = data.get('roots', {})
|
|
return bookmarks
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error: {e}")
|
|
return None
|
|
|
|
|
|
# todo: move this to the main.py
|
|
def get_extensions_installed_from_shop(env, config: Dict[str, str]):
|
|
"""Find the Chrome extensions directory based on the operating system."""
|
|
os_type = env.vm_platform
|
|
if os_type == 'Windows':
|
|
chrome_extension_dir = env.controller.execute_python_command(
|
|
"""os.path.expanduser('~') + '\\AppData\\Local\\Google\\Chrome\\User Data\\Default\\Extensions\\'""")[
|
|
'output'].strip()
|
|
elif os_type == 'Darwin': # macOS
|
|
chrome_extension_dir = env.controller.execute_python_command(
|
|
"""os.path.expanduser('~') + '/Library/Application Support/Google/Chrome/Default/Extensions/'""")[
|
|
'output'].strip()
|
|
elif os_type == 'Linux':
|
|
chrome_extension_dir = env.controller.execute_python_command(
|
|
"""os.path.expanduser('~') + '/.config/google-chrome/Default/Extensions/'""")['output'].strip()
|
|
else:
|
|
raise Exception('Unsupported operating system')
|
|
|
|
manifests = []
|
|
for extension_id in os.listdir(chrome_extension_dir):
|
|
extension_path = os.path.join(chrome_extension_dir, extension_id)
|
|
if os.path.isdir(extension_path):
|
|
# Iterate through version-named subdirectories
|
|
for version_dir in os.listdir(extension_path):
|
|
version_path = os.path.join(extension_path, version_dir)
|
|
manifest_path = os.path.join(version_path, 'manifest.json')
|
|
if os.path.isfile(manifest_path):
|
|
with open(manifest_path, 'r') as file:
|
|
try:
|
|
manifest = json.load(file)
|
|
manifests.append(manifest)
|
|
except json.JSONDecodeError:
|
|
logger.error(f"Error reading {manifest_path}")
|
|
return manifests
|
|
|
|
|
|
# The following ones require Playwright to be installed on the target machine, and the chrome needs to be pre-config on
|
|
# port info to allow remote debugging, see README.md for details
|
|
|
|
def get_open_tabs_info(env, config: Dict[str, str]):
|
|
host = env.vm_ip
|
|
port = 9222 # fixme: this port is hard-coded, need to be changed from config file
|
|
|
|
remote_debugging_url = f"http://{host}:{port}"
|
|
with sync_playwright() as p:
|
|
# connect to remote Chrome instance
|
|
browser = p.chromium.connect_over_cdp(remote_debugging_url)
|
|
|
|
tabs_info = []
|
|
for context in browser.contexts:
|
|
for page in context.pages:
|
|
title = page.title()
|
|
url = page.url
|
|
tabs_info.append({'title': title, 'url': url})
|
|
|
|
browser.close()
|
|
return tabs_info
|
|
|
|
|
|
def get_active_tab_info(env, config: Dict[str, str]):
|
|
host = env.vm_ip
|
|
port = 9222 # fixme: this port is hard-coded, need to be changed from config file
|
|
|
|
remote_debugging_url = f"http://{host}:{port}"
|
|
with sync_playwright() as p:
|
|
# connect to remote Chrome instance
|
|
browser = p.chromium.connect_over_cdp(remote_debugging_url)
|
|
|
|
active_tab_info = {}
|
|
for context in browser.contexts:
|
|
for page in context.pages:
|
|
if page.is_visible("body"): # check the visibility of the page body to determine the active status
|
|
active_tab_info = {
|
|
'title': page.title(),
|
|
'url': page.url,
|
|
'content': page.content() # get the HTML content of the page
|
|
}
|
|
break
|
|
if active_tab_info:
|
|
break
|
|
|
|
browser.close()
|
|
return active_tab_info
|