import json import os import random import time import traceback import requests import requests.utils from . import config, devices # ====== SYNC METHODS ====== # def sync_device_features(self, login=False): data = { "id": self.uuid, "server_config_retrieval": "1", "experiments": config.LOGIN_EXPERIMENTS, } if login is False: data["_uuid"] = self.uuid data["_uid"] = self.user_id data["_csrftoken"] = self.token data = json.dumps(data) self.last_experiments = time.time() return self.send_request( "qe/sync/", data, login=login, headers={"X-DEVICE-ID": self.uuid} ) def sync_launcher(self, login=False): data = { "id": self.uuid, "server_config_retrieval": "1", } if login is False: data["_uid"] = self.user_id data["_uuid"] = self.uuid data["_csrftoken"] = self.token data = json.dumps(data) return self.send_request("launcher/sync/", data, login=login) def sync_user_features(self): data = self.default_data data["id"] = self.uuid data["experiments"] = config.EXPERIMENTS data = json.dumps(data) self.last_experiments = time.time() return self.send_request("qe/sync/", data, headers={"X-DEVICE-ID": self.uuid}) # "android_device_id":"android-f14b9731e4869eb", # "phone_id":"b4bd7978-ca2b-4ea0-a728-deb4180bd6ca", # "usages":"[\"account_recovery_omnibox\"]", # "_csrftoken":"9LZXBXXOztxNmg3h1r4gNzX5ohoOeBkI", # "device_id":"70db6a72-2663-48da-96f5-123edff1d458" def get_prefill_candidates(self, login=False): data = { "android_device_id": self.device_id, "phone_id": self.phone_id, "usages": '["account_recovery_omnibox"]', "device_id": self.device_id, } if login is False: data["_csrftoken"] = self.token data = json.dumps(data) return self.send_request("accounts/get_prefill_candidates/", data, login=login) def get_account_family(self): return self.send_request("multiple_accounts/get_account_family/") def get_zr_token_result(self): url = ( "zr/token/result/?device_id={rank_token}" "&token_hash=&custom_device_id={custom_device_id}&fetch_reason=token_expired" ) url = url.format(rank_token=self.device_id, custom_device_id=self.device_id) return self.send_request(url) def banyan(self): url = "banyan/banyan/?views=['story_share_sheet','threads_people_picker','group_stories_share_sheet','reshare_share_sheet']" return self.send_request(url) def igtv_browse_feed(self): url = "igtv/browse_feed/?prefetch=1" return self.send_request(url) def creatives_ar_class(self): data = { "_csrftoken": self.token, "_uuid": self.uuid, } data = json.dumps(data) return self.send_request("creatives/ar_class/", data) # ====== LOGIN/PRE FLOWS METHODS ====== # def pre_login_flow(self): self.logger.info("Not yet logged in starting: PRE-LOGIN FLOW!") # /api/v1/accounts/get_prefill_candidates self.get_prefill_candidates(True) # /api/v1/qe/sync (server_config_retrieval) self.sync_device_features(True) # /api/v1/launcher/sync/ (server_config_retrieval) self.sync_launcher(True) # /api/v1/accounts/contact_point_prefill/ self.set_contact_point_prefill("prefill") # /api/v1/accounts/get_prefill_candidates/ # self.get_prefill_candidates(True) # /api/v1/accounts/contact_point_prefill/ # self.set_contact_point_prefill("prefill") # /api/v1/qe/sync/ (server_config_retrieval) # self.sync_device_features(True) # /api/v1/launcher/sync/ (server_config_retrieval) # self.sync_launcher(True) # DO NOT MOVE ANY OF THE ENDPOINTS THEYRE IN THE CORRECT ORDER def login_flow(self, just_logged_in=False, app_refresh_interval=1800): self.last_experiments = time.time() self.logger.info("LOGIN FLOW! Just logged-in: {}".format(just_logged_in)) check_flow = [] if just_logged_in: try: # SYNC # /api/v1/qe/sync/ (server_config_retrieval) check_flow.append(self.sync_device_features()) # /api/v1/launcher/sync/ (server_config_retrieval) check_flow.append(self.sync_launcher(False)) # /api/v1/zr/token/result/?device_id=android-f14b9731e4869eb&token_hash=&custom_device_id=f3119c98-5663-4c47-95b5-a63b140a2b62&fetch_reason=token_expired check_flow.append(self.get_zr_token_result()) # /api/v1/multiple_accounts/get_account_family/ check_flow.append(self.get_account_family()) # /api/v1/qe/sync/ (server_config_retrieval) check_flow.append(self.sync_device_features()) # /api/v1/banyan/banyan/?views=%5B%22story_share_sheet%22%2C%22threads_people_picker%22%2C%22group_stories_share_sheet%22%2C%22reshare_share_sheet%22%5D # error 400: {"message": "Bad request", "status": "fail"} # check_flow.append(self.banyan()) # /api/v1/igtv/browse_feed/?prefetch=1 check_flow.append(self.igtv_browse_feed()) # /api/v1/creatives/ar_class/ (_csrftoken also add _uuid) check_flow.append(self.creatives_ar_class()) # /api/v1/feed/reels_tray/ (supported_capabilities_new + reason=cold_start + _csrftoken + _uuid) check_flow.append(self.get_reels_tray_feed(reason="cold_start")) # /api/v1/feed/timeline/ (feed_view_info + phone_id + battery_level + timezone_offset + _csrftoken + device_id + request_id + is_pull_to_refresh=0 + _uuid + is_charging=1 + will_sound_on=0 + seesion_id + bloks_versioning_id) check_flow.append(self.get_timeline_feed()) # /api/v1/feed/reels_media/ # error 400: {"message": "Invalid reel id list", "status": "fail"} # check_flow.append(self.get_reels_media()) # /api/v1/push/register/ (device_type=android_mqtt + is_main_push_channel=true + device_sub_type=2 + device_toke + _csrftoken + guid + _uuid + users + family_device_id) # device_type=android_mqtt&is_main_push_channel=true&device_sub_type=2&device_token={"k":"eyJwbiI6ImNvbS5pbnN0YWdyYW0uYW5kcm9pZCIsImRpIjoiNzhlNGMxNmQtN2YzNC00NDlkLTg4OWMtMTAwZDg5OTU0NDJhIiwiYWkiOjU2NzMxMDIwMzQxNTA1MiwiY2siOiIxNjgzNTY3Mzg0NjQyOTQifQ==","v":0,"t":"fbns-b64"}&_csrftoken=mmdoMLXFQEzt2w5xLbfm0FTs7gIgqAlc&guid=f87b5e9f-0663-42f8-9213-ec72cb49c961&_uuid=f87b5e9f-0663-42f8-9213-ec72cb49c961&users=3149016955&family_device_id=9d9aa0f0-40fe-4524-a920-9910f45ba18d # error 400: {"message": "no token provided", "status": "fail"} # check_flow.append(self.push_register()) # /api/v1/feed/reels_media/ # error 400: {"message": "Invalid reel id list", "status": "fail"} # check_flow.append(self.get_reels_media()) # /api/v1/media/blocked/ check_flow.append(self.media_blocked()) # /api/v1/news/inbox/ check_flow.append(self.get_news_inbox()) # /api/v1/loom/fetch_config/ check_flow.append(self.get_loom_fetch_config()) # /api/v1/scores/bootstrap/users/?surfaces=%5B%22autocomplete_user_list%22%2C%22coefficient_besties_list_ranking%22%2C%22coefficient_rank_recipient_user_suggestion%22%2C%22coefficient_ios_section_test_bootstrap_ranking%22%2C%22coefficient_direct_recipients_ranking_variant_2%22%5D check_flow.append(self.get_scores_bootstrap()) # /api/v1/business/eligibility/get_monetization_products_eligibility_data/?product_types=branded_content check_flow.append(self.get_monetization_products_eligibility_data()) # /api/v1/business/branded_content/should_require_professional_account/ check_flow.append(self.get_business_branded_content()) # /api/v1/linked_accounts/get_linkage_status/ check_flow.append(self.get_linked_accounts()) # /api/v1/locations/request_country/ check_flow.append(self.get_request_country()) # /api/v1/qp/get_cooldowns/?signed_body=a7c6081ee2ae5b41a1475f83b6dbc8f1130c67d472f69748221468e1621823b5.%7B%7D&ig_sig_key_version=4 check_flow.append(self.get_cooldowns()) # /api/v1/users/arlink_download_info/?version_override=2.2.1 check_flow.append(self.arlink_download_info()) # push register # error 400: {"message": "no token provided", "status": "fail"} # check_flow.append(self.push_register()) # /api/v1/users/self.user_id/info/ check_flow.append(self.get_username_info(self.user_id)) # /api/v1/notifications/store_client_push_permissions/ # error 400: {"message": "missing param", "status": "fail"} # check_flow.append(self.log_resurrect_attribution()) # /api/v1/accounts/process_contact_point_signals # error 400: {"message": "", "status": "fail"} # check_flow.append(self.process_contact_point_signals()) # creatives/write_supported_capabilities # error 500: unknown # check_flow.append(self.write_supported_capabilities()) check_flow.append(self.get_presence()) check_flow.append(self.get_direct_v2_inbox()) check_flow.append(self.get_direct_v2_inbox2()) check_flow.append(self.topical_explore()) check_flow.append(self.notification_badge()) check_flow.append(self.batch_fetch()) # TODO add facebook_ota # /api/v1/facebook_ota/?fields=update%7Bdownload_uri%2Cdownload_uri_delta_base%2Cversion_code_delta_base%2Cdownload_uri_delta%2Cfallback_to_full_update%2Cfile_size_delta%2Cversion_code%2Cpublished_date%2Cfile_size%2Cota_bundle_type%2Cresources_checksum%2Callowed_networks%2Crelease_id%7D&custom_user_id=3149016955&signed_body=656adcfd879d775324e9c1668534f80a999801c7780f16cc720d7970941195de.&ig_sig_key_version=4&version_code=195435566&version_name=126.0.0.25.121&custom_app_id=124024574287414&custom_device_id=f87b5e9f-0663-42f8-9213-ec72cb49c961 HTTP/1.1 # check_flow.append(self.facebook_ota()) except Exception as e: self.logger.error( "Exception raised: {}\n{}".format(e, traceback.format_exc()) ) return False else: try: pull_to_refresh = random.randint(1, 100) % 2 == 0 check_flow.append( self.get_timeline_feed( options=["is_pull_to_refresh"] if pull_to_refresh is True else [] ) ) # Random pull_to_refresh :) check_flow.append( self.get_reels_tray_feed( reason="pull_to_refresh" if pull_to_refresh is True else "cold_start" ) ) is_session_expired = (time.time() - self.last_login) > app_refresh_interval if is_session_expired: self.last_login = time.time() self.client_session_id = self.generate_UUID(uuid_type=True) # getBootstrapUsers() ... check_flow.append(self.get_ranked_recipients("reshare", True)) check_flow.append(self.get_ranked_recipients("save", True)) check_flow.append(self.get_inbox_v2()) check_flow.append(self.get_presence()) check_flow.append(self.get_recent_activity()) check_flow.append(self.get_profile_notice()) check_flow.append(self.explore(False)) if (time.time() - self.last_experiments) > 7200: check_flow.append(self.sync_user_features()) check_flow.append(self.sync_device_features()) except Exception as e: self.logger.error( "Error loginin, exception raised: {}\n{}".format( e, traceback.format_exc() ) ) return False self.save_uuid_and_cookie() return False if False in check_flow else True # ====== DEVICE / CLIENT_ID / PHONE_ID AND OTHER VALUES (uuids) ====== # def set_device(self): self.device_settings = devices.DEVICES[self.device] self.user_agent = config.USER_AGENT_BASE.format(**self.device_settings) def generate_all_uuids(self): self.phone_id = self.generate_UUID(uuid_type=True) self.uuid = self.generate_UUID(uuid_type=True) self.client_session_id = self.generate_UUID(uuid_type=True) self.advertising_id = self.generate_UUID(uuid_type=True) self.device_id = self.generate_device_id( self.get_seed(self.username, self.password) ) def reinstall_app_simulation(self): self.logger.info("Reinstall app simulation, generating new `phone_id`...") self.phone_id = self.generate_UUID(uuid_type=True) self.save_uuid_and_cookie() self.logger.info("New phone_id: {}".format(self.phone_id)) def change_device_simulation(self): self.logger.info("Change device simulation") self.reinstall_app_simulation() self.logger.info("Generating new `android_device_id`...") self.device_id = self.generate_device_id( self.get_seed(self.generate_UUID(uuid_type=True)) ) self.save_uuid_and_cookie() self.logger.info("New android_device_id: {}".format(self.device_id)) def load_uuid_and_cookie(self, load_uuid=True, load_cookie=True): if self.cookie_fname is None: fname = "{}_uuid_and_cookie.json".format(self.username) self.cookie_fname = os.path.join(self.base_path, fname) print(os.path.join(self.base_path, fname)) if os.path.isfile(self.cookie_fname) is False: return False with open(self.cookie_fname, "r") as f: data = json.load(f) if "cookie" in data: self.last_login = data["timing_value"]["last_login"] self.last_experiments = data["timing_value"]["last_experiments"] if load_cookie: self.logger.debug("Loading cookies") self.session.cookies = requests.utils.cookiejar_from_dict( data["cookie"] ) cookie_username = self.cookie_dict["ds_user"] assert cookie_username == self.username.lower() self.cookie_dict["urlgen"] if load_uuid: self.logger.debug("Loading uuids") self.phone_id = data["uuids"]["phone_id"] self.uuid = data["uuids"]["uuid"] self.client_session_id = data["uuids"]["client_session_id"] self.advertising_id = data["uuids"]["advertising_id"] self.device_id = data["uuids"]["device_id"] self.device_settings = data["device_settings"] self.user_agent = data["user_agent"] msg = ( "Recovery from {}: COOKIE {} - UUIDs {} - TIMING, DEVICE " "and ...\n- user-agent={}\n- phone_id={}\n- uuid={}\n- " "client_session_id={}\n- device_id={}" ) self.logger.info( msg.format( self.cookie_fname, load_cookie, load_uuid, self.user_agent, self.phone_id, self.uuid, self.client_session_id, self.device_id, ) ) else: self.logger.info( "The cookie seems to be the with the older structure. " "Load and init again all uuids" ) self.session.cookies = requests.utils.cookiejar_from_dict(data) self.last_login = time.time() self.last_experiments = time.time() cookie_username = self.cookie_dict["ds_user"] assert cookie_username == self.username self.set_device() self.generate_all_uuids() self.is_logged_in = True return True def save_uuid_and_cookie(self): if self.cookie_fname is None: fname = "{}_uuid_and_cookie.json".format(self.username) self.cookie_fname = os.path.join(self.base_path, fname) data = { "uuids": { "phone_id": self.phone_id, "uuid": self.uuid, "client_session_id": self.client_session_id, "advertising_id": self.advertising_id, "device_id": self.device_id, }, "cookie": requests.utils.dict_from_cookiejar(self.session.cookies), "timing_value": { "last_login": self.last_login, "last_experiments": self.last_experiments, }, "device_settings": self.device_settings, "user_agent": self.user_agent, } with open(self.cookie_fname, "w") as f: json.dump(data, f)