Merge branch 'main' into xiaochuanli/addChromeExtensions

This commit is contained in:
Tianbao Xie
2024-02-25 00:45:17 +08:00
committed by GitHub
152 changed files with 7174 additions and 420 deletions

View File

@@ -326,6 +326,36 @@ def get_extensions_installed_from_shop(env, config: Dict[str, str]):
# 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_page_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
url = config["url"]
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)
page = browser.contexts[0].new_page()
page.goto(url)
try:
# Wait for the page to finish loading, this prevents the "execution context was destroyed" issue
page.wait_for_load_state('load') # Wait for the 'load' event to complete
title = page.title()
url = page.url
page_info = {'title': title, 'url': url, 'content': page.content()}
except TimeoutError:
# If page loading times out, catch the exception and store the current information in the list
page_info = {'title': 'Load timeout', 'url': page.url, 'content': page.content()}
except Exception as e:
# Catch other potential exceptions that might occur while reading the page title
print(f'Error: {e}')
page_info = {'title': 'Error encountered', 'url': page.url, 'content': page.content()}
browser.close()
return page_info
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
@@ -551,9 +581,9 @@ def get_googledrive_file(env, config: Dict[str, Any]) -> str:
for q in _query:
search = f'( {q} ) and "{parent_id}" in parents'
filelist: GoogleDriveFileList = drive.ListFile({'q': search}).GetList()
if len(filelist) == 0: # target file not found
if len(filelist) == 0: # target file not found
return None
file: GoogleDriveFile = filelist[0] # HACK: if multiple candidates, just use the first one
file: GoogleDriveFile = filelist[0] # HACK: if multiple candidates, just use the first one
parent_id = file['id']
file.GetContentFile(_path, mimetype=file['mimeType'])
@@ -565,8 +595,9 @@ def get_googledrive_file(env, config: Dict[str, Any]) -> str:
if 'query' in config:
return get_single_file(config['query'], os.path.join(env.cache_dir, config['dest']))
elif 'path' in config:
query = [f"title = '{fp}' and mimeType = 'application/vnd.google-apps.folder' and trashed = false" if idx < len(config['path']) - 1
else f"title = '{fp}' and trashed = false" for idx, fp in enumerate(config['path'])]
query = [f"title = '{fp}' and mimeType = 'application/vnd.google-apps.folder' and trashed = false" if idx < len(
config['path']) - 1
else f"title = '{fp}' and trashed = false" for idx, fp in enumerate(config['path'])]
return get_single_file(query, os.path.join(env.cache_dir, config['dest']))
elif 'query_list' in config:
_path_list = []
@@ -575,12 +606,14 @@ def get_googledrive_file(env, config: Dict[str, Any]) -> str:
dest = config['dest'][idx]
_path_list.append(get_single_file(query, os.path.join(env.cache_dir, dest)))
return _path_list
else: # path_list in config
else: # path_list in config
_path_list = []
assert len(config['path_list']) == len(config['dest'])
for idx, path in enumerate(config['path_list']):
query = [f"title = '{fp}' and mimeType = 'application/vnd.google-apps.folder' and trashed = false" if jdx < len(path) - 1
else f"title = '{fp}' and trashed = false" for jdx, fp in enumerate(path)]
query = [
f"title = '{fp}' and mimeType = 'application/vnd.google-apps.folder' and trashed = false" if jdx < len(
path) - 1
else f"title = '{fp}' and trashed = false" for jdx, fp in enumerate(path)]
dest = config['dest'][idx]
_path_list.append(get_single_file(query, os.path.join(env.cache_dir, dest)))
return _path_list
@@ -596,12 +629,12 @@ def get_enable_do_not_track(env, config: Dict[str, str]):
"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()
preference_file_path = env.controller.execute_python_command(
"import os; print(os.path.join(os.getenv('HOME'), 'snap/chromium/common/chromium/Default/Preferences'))")[
"import os; print(os.path.join(os.getenv('HOME'), '.config/google-chrome/Default/Preferences'))")[
'output'].strip()
# preference_file_path = env.controller.execute_python_command(
# "import os; print(os.path.join(os.getenv('HOME'), 'snap/chromium/common/chromium/Default/Preferences'))")[
# 'output'].strip()
else:
raise Exception('Unsupported operating system')
@@ -609,11 +642,11 @@ def get_enable_do_not_track(env, config: Dict[str, str]):
content = env.controller.get_file(preference_file_path)
data = json.loads(content)
if_enable_do_not_track = data.get('enable_do_not_track', {}) # bool
if_enable_do_not_track = data.get('enable_do_not_track', {}) # bool
return "true" if if_enable_do_not_track else "false"
except Exception as e:
logger.error(f"Error: {e}")
return "Google"
return "false"
def get_enable_enhanced_safety_browsing(env, config: Dict[str, str]):
@@ -626,12 +659,12 @@ def get_enable_enhanced_safety_browsing(env, config: Dict[str, str]):
"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()
preference_file_path = env.controller.execute_python_command(
"import os; print(os.path.join(os.getenv('HOME'), 'snap/chromium/common/chromium/Default/Preferences'))")[
"import os; print(os.path.join(os.getenv('HOME'), '.config/google-chrome/Default/Preferences'))")[
'output'].strip()
# preference_file_path = env.controller.execute_python_command(
# "import os; print(os.path.join(os.getenv('HOME'), 'snap/chromium/common/chromium/Default/Preferences'))")[
# 'output'].strip()
else:
raise Exception('Unsupported operating system')
@@ -639,7 +672,7 @@ def get_enable_enhanced_safety_browsing(env, config: Dict[str, str]):
content = env.controller.get_file(preference_file_path)
data = json.loads(content)
if_enable_do_not_track = data.get('safebrowsing', {}).get('enhanced', {}) # bool
if_enable_do_not_track = data.get('safebrowsing', {}).get('enhanced', {}) # bool
return "true" if if_enable_do_not_track else "false"
except Exception as e:
logger.error(f"Error: {e}")
@@ -656,12 +689,12 @@ def get_new_startup_page(env, config: Dict[str, str]):
"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()
preference_file_path = env.controller.execute_python_command(
"import os; print(os.path.join(os.getenv('HOME'), 'snap/chromium/common/chromium/Default/Preferences'))")[
"import os; print(os.path.join(os.getenv('HOME'), '.config/google-chrome/Default/Preferences'))")[
'output'].strip()
# preference_file_path = env.controller.execute_python_command(
# "import os; print(os.path.join(os.getenv('HOME'), 'snap/chromium/common/chromium/Default/Preferences'))")[
# 'output'].strip()
else:
raise Exception('Unsupported operating system')
@@ -674,7 +707,7 @@ def get_new_startup_page(env, config: Dict[str, str]):
if "session" not in data.keys():
return "true"
else:
if_enable_do_not_track = data.get('session', {}).get('restore_on_startup', {}) # int, need to be 5
if_enable_do_not_track = data.get('session', {}).get('restore_on_startup', {}) # int, need to be 5
return "true" if if_enable_do_not_track == 5 else "false"
except Exception as e:
logger.error(f"Error: {e}")
@@ -691,12 +724,12 @@ def get_find_unpacked_extension_path(env, config: Dict[str, str]):
"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()
preference_file_path = env.controller.execute_python_command(
"import os; print(os.path.join(os.getenv('HOME'), 'snap/chromium/common/chromium/Default/Preferences'))")[
"import os; print(os.path.join(os.getenv('HOME'), '.config/google-chrome/Default/Preferences'))")[
'output'].strip()
# preference_file_path = env.controller.execute_python_command(
# "import os; print(os.path.join(os.getenv('HOME'), 'snap/chromium/common/chromium/Default/Preferences'))")[
# 'output'].strip()
else:
raise Exception('Unsupported operating system')