import datetime import logging import os.path import time import urllib.request from copy import deepcopy from playwright.sync_api import sync_playwright, Request, Page from configs.idp_rules import idp_rules from modules.browser.browser import PlaywrightBrowser, PlaywrightHelper from modules.helper.tmp import TmpHelper from modules.helper.url import URLHelper logger = logging.getLogger(__name__) idp_rules_with_generic = deepcopy(idp_rules) idp_rules_with_generic["FACEBOOK"]["passive_login_request_rule"] = { "domain": "facebook\\.com$", "path": "/x/oauth/", "params": [ { "name": "^(client_id|app_id)$", "value": ".*" } ] } idp_rules_with_generic["GENERIC"] = { "login_request_rule": { "domain": ".*", "path": ".*", "params": [ {"name": "^(client_id|clientid|app_id|appid)$", "value": ".*"}, {"name": "^(redirect_uri|redirecturi)$", "value": ".*"} ]}, "passive_login_request_rule": {} } user_actions = ["PageDown", "PageUp", "ArrowRight", "ArrowLeft", "ArrowDown", "ArrowUp"] cookie_block_checkbox_ids = { "functional_cookies": "func_checkbox", "analytical_cookies": "anal_checkbox", "advertising_and_tracking_cookies": "advert_checkbox" } class Privacy: def __init__(self, config: dict): self.page = config['page'] if config['page'].startswith('http') else "https://" + config['page'] self.browser_config = config['browser_config'] self.rest_time_on_page = config['rest_time_on_page'] self.simulate_interactions = config['simulate_interactions'] self.store_screenshot_if_request_found = config['artifacts_config']["store_screenshot_if_request_found"] self.store_screenshot_if_no_request_found = config['artifacts_config']["store_screenshot_if_no_request_found"] self.store_har_if_request_found = config['artifacts_config']["store_har_if_request_found"] self.store_har_if_no_request_found = config['artifacts_config']["store_har_if_no_request_found"] self.cookie_banner_procedure_config = config['cookie_banner_procedure'] self.found_sso_requests = [] self.screenshot = None self.thrown_exception = None self.user_action_performed = False @staticmethod def schedule(scan_config: dict, analysis_config: dict): if scan_config['scan_type'] == 'single': c = deepcopy(analysis_config) c["page"] = scan_config["page"] yield c elif scan_config["scan_type"] == "range": with TmpHelper().tmp_dir() as tmp_dir: downloaded_file = tmp_dir + scan_config['list_id'] os.makedirs(tmp_dir, exist_ok=True) file_url = f"https://tranco-list.eu/download/{scan_config['list_id']}/{scan_config['limit']}" logger.info( f"Downloading tranco list with id {scan_config['list_id']} ({file_url}) to {downloaded_file}") urllib.request.urlretrieve(file_url, downloaded_file) with open(downloaded_file) as tranco_list: for line in tranco_list.readlines(): entry = line.split(",") if int(scan_config["limit"]) >= int(entry[0]) >= int(scan_config["offset"]): c = deepcopy(analysis_config) c["tranco_id"] = int(entry[0]) c["page"] = entry[1].replace("\n", "") yield c def interceptor(self, req: Request): for idp in idp_rules_with_generic: if (any(found_req["idp"] == idp for found_req in self.found_sso_requests) or (idp == "GENERIC" and len(self.found_sso_requests) > 0)): continue lreq_rule = idp_rules_with_generic[idp]["login_request_rule"] plreq_rule = idp_rules_with_generic[idp]["passive_login_request_rule"] if lreq_rule and URLHelper.match_url(req.url, lreq_rule["domain"], lreq_rule["path"], lreq_rule["params"]): logger.info("Found a login request for %s", self.page) self.found_sso_requests.append( {"idp": idp, "lreq": req.url, "user_action_performed": self.user_action_performed}) elif plreq_rule and URLHelper.match_url(req.url, plreq_rule["domain"], plreq_rule["path"], plreq_rule["params"]): logger.info("Found a **passive** login request for %s", self.page) self.found_sso_requests.append( {"idp": idp, "lreq": req.url, "user_action_performed": self.user_action_performed}) def perform_simulated_user_action(self, page: Page): logger.info("Performing simulated user actions") self.user_action_performed = True page.mouse.move(23, 55) time.sleep(0.1) page.mouse.move(35, 121) time.sleep(0.1) for i in range(20): page.mouse.wheel(0, 10) time.sleep(0.01) for i in range(20): page.mouse.wheel(0, -10) time.sleep(0.01) time.sleep(0.25) for ua in user_actions: page.keyboard.press(ua) time.sleep(0.25) time.sleep(0.5) def setup_cookie_blocker_extension(self, context): logger.info("Setting up cookie blocker extension") page = context.new_page() page.goto("chrome-extension://fbhiolckidkciamgcobkokpelckgnnol/options/cookieblock_options.html") el_func_cookies_cb = page.locator(f"#%s" % cookie_block_checkbox_ids['functional_cookies']) el_analytical_cookies_cb = page.locator(f"#%s" % cookie_block_checkbox_ids['analytical_cookies']) el_advertising_and_tracking_cookies_cb = page.locator( f"#%s" % cookie_block_checkbox_ids['advertising_and_tracking_cookies']) if el_func_cookies_cb.is_checked() != self.cookie_banner_procedure_config['accept_functional_cookies']: el_func_cookies_cb.click() if el_analytical_cookies_cb.is_checked() != self.cookie_banner_procedure_config['accept_analytical_cookies']: el_analytical_cookies_cb.click() if el_advertising_and_tracking_cookies_cb.is_checked() != self.cookie_banner_procedure_config[ 'accept_advertising_and_tracking_cookies']: el_advertising_and_tracking_cookies_cb.click() def start(self): start_time = datetime.datetime.now() logger.info("Starting privacy analysis for %s", self.page) har = None store_screenshot = False if self.cookie_banner_procedure_config["execute_cookie_banner_actions"] and not any( "cookieblock" in ext for ext in self.browser_config["extensions"]): logger.warning("The execution of cookie banner procedure was set to true but the cookie block extension " "is not present in the browser config. This will most likely cause the execution of this " "task to fail!") elif not self.cookie_banner_procedure_config["execute_cookie_banner_actions"] and any( "cookieblock" in ext for ext in self.browser_config["extensions"]): logger.warning("The execution of cookie banner procedure was set to false but the cookie block extension " "is present in the browser config. The analysis will be performed as if execution was set " "to true with default config (accept functional cookies, decline others)") with TmpHelper.tmp_dir() as pdir, TmpHelper.tmp_file() as har_file, sync_playwright() as pw: context, page = PlaywrightBrowser.instance(pw, self.browser_config, pdir, har_file if ( self.store_har_if_request_found or self.store_har_if_no_request_found) else None) context.on("request", self.interceptor) try: if self.cookie_banner_procedure_config['execute_cookie_banner_actions']: self.setup_cookie_blocker_extension(context) PlaywrightHelper.close_all_other_pages(page) PlaywrightHelper.navigate(page, self.page) logger.info("Loaded page, waiting for %s seconds...", self.rest_time_on_page) time.sleep(self.rest_time_on_page) if self.simulate_interactions: self.perform_simulated_user_action(page) else: logger.debug("Simulating interactions has been disabled") store_screenshot = (self.store_screenshot_if_request_found and len(self.found_sso_requests) > 0) or ( self.store_screenshot_if_no_request_found and len(self.found_sso_requests) == 0) if store_screenshot: self.screenshot = PlaywrightHelper.take_screenshot(page) except Exception as e: logger.info("Error while performing analysis: " + str(e)) self.thrown_exception = str(e) store_har = (self.store_har_if_request_found and len(self.found_sso_requests) > 0) or ( self.store_har_if_no_request_found and len(self.found_sso_requests) == 0) PlaywrightHelper.close_context(context) if store_har: har = PlaywrightHelper.take_har(har_file) duration = datetime.datetime.now() - start_time logger.info("Finished task in %s seconds" % duration.seconds) return_value = {"page": self.page, "found_sso_requests": self.found_sso_requests, "thrown_exception": self.thrown_exception, "duration": duration.seconds, "screenshot_zlib_compressed": self.screenshot if store_screenshot else None, "har_zlib_compressed": har if store_har else None} return return_value