Implement account-related HTTP requests

mauigpapi/errors/__init__.py

@@ -0,0 +1,8 @@
+from .base import IGError
+from .mqtt import IGMQTTError, NotLoggedIn, NotConnected
+from .state import IGUserIDNotFoundError, IGCookieNotFoundError, IGNoCheckpointError
+from .response import (IGResponseError, IGActionSpamError, IGNotFoundError, IGRateLimitError,
+                       IGCheckpointError, IGUserHasLoggedOutError, IGLoginRequiredError,
+                       IGPrivateUserError, IGSentryBlockError, IGInactiveUserError, IGLoginError,
+                       IGLoginTwoFactorRequiredError, IGLoginBadPasswordError,
+                       IGLoginInvalidUserError)

mauigpapi/errors/base.py

@@ -0,0 +1,18 @@
+# mautrix-instagram - A Matrix-Instagram puppeting bridge.
+# Copyright (C) 2020 Tulir Asokan
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Affero General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <https://www.gnu.org/licenses/>.
+
+class IGError(Exception):
+    pass

mauigpapi/errors/mqtt.py

@@ -0,0 +1,28 @@
+# mautrix-instagram - A Matrix-Instagram puppeting bridge.
+# Copyright (C) 2020 Tulir Asokan
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Affero General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <https://www.gnu.org/licenses/>.
+from .base import IGError
+
+
+class IGMQTTError(IGError):
+    pass
+
+
+class NotLoggedIn(IGMQTTError):
+    pass
+
+
+class NotConnected(IGMQTTError):
+    pass

mauigpapi/errors/response.py

@@ -0,0 +1,101 @@
+# mautrix-instagram - A Matrix-Instagram puppeting bridge.
+# Copyright (C) 2020 Tulir Asokan
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Affero General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <https://www.gnu.org/licenses/>.
+from typing import Optional, get_type_hints
+
+from aiohttp import ClientResponse
+from mautrix.types import JSON, Serializable
+
+from .base import IGError
+from ..types import SpamResponse, CheckpointResponse, LoginRequiredResponse, LoginErrorResponse
+
+
+class IGResponseError(IGError):
+    response: ClientResponse
+
+    def __init__(self, response: ClientResponse, json: JSON) -> None:
+        prefix = f"Request {response.request_info.method} {response.request_info.url.path} failed"
+        message = f"HTTP {response.status}"
+        self.response = response
+        if "message" in json:
+            message = json["message"]
+        type_hint = get_type_hints(type(self)).get("body", JSON)
+        if type_hint is not JSON and isinstance(type_hint, Serializable):
+            self.body = type_hint.deserialize(json)
+        super().__init__(f"{prefix}: {self._message_override or message}")
+
+    @property
+    def _message_override(self) -> Optional[str]:
+        return None
+
+
+class IGActionSpamError(IGResponseError):
+    body: SpamResponse
+
+    @property
+    def _message(self) -> str:
+        return f"HTTP {self.body.message}"
+
+
+class IGNotFoundError(IGResponseError):
+    pass
+
+
+class IGRateLimitError(IGResponseError):
+    pass
+
+
+class IGCheckpointError(IGResponseError):
+    body: CheckpointResponse
+
+    @property
+    def url(self) -> str:
+        return self.body.challenge.api_path
+
+
+class IGUserHasLoggedOutError(IGResponseError):
+    body: LoginRequiredResponse
+
+
+class IGLoginRequiredError(IGResponseError):
+    body: LoginRequiredResponse
+
+
+class IGPrivateUserError(IGResponseError):
+    pass
+
+
+class IGSentryBlockError(IGResponseError):
+    pass
+
+
+class IGInactiveUserError(IGResponseError):
+    pass
+
+
+class IGLoginError(IGResponseError):
+    body: LoginErrorResponse
+
+
+class IGLoginTwoFactorRequiredError(IGLoginError):
+    pass
+
+
+class IGLoginBadPasswordError(IGLoginError):
+    pass
+
+
+class IGLoginInvalidUserError(IGLoginError):
+    pass

mauigpapi/errors.py → mauigpapi/errors/state.py

@@ -13,17 +13,7 @@
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <https://www.gnu.org/licenses/>.
-
-class IGError(Exception):
-    pass
-
-
-class NotLoggedIn(IGError):
-    pass
-
-
-class NotConnected(IGError):
-    pass
+from .base import IGError
 
 
 class IGUserIDNotFoundError(IGError):

mauigpapi/http/account.py

@@ -0,0 +1,145 @@
+# mautrix-instagram - A Matrix-Instagram puppeting bridge.
+# Copyright (C) 2020 Tulir Asokan
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Affero General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <https://www.gnu.org/licenses/>.
+from typing import Optional, Type, TypeVar
+import base64
+import struct
+import time
+import json
+import io
+
+from ..types import CurrentUserResponse
+from .base import BaseAndroidAPI
+
+
+T = TypeVar('T')
+
+
+class AccountAPI(BaseAndroidAPI):
+    async def current_user(self) -> CurrentUserResponse:
+        url = (self.url / "api/v1/accounts/current_user/").with_query({"edit": "true"})
+        resp = await self.http.get(url)
+        return CurrentUserResponse.deserialize(await self.handle_response(resp))
+
+    async def set_biography(self, text: str) -> CurrentUserResponse:
+        # TODO entities?
+        return await self.__command("set_biography", device_id=self.state.device.id, raw_text=text)
+
+    async def set_profile_picture(self, upload_id: str) -> CurrentUserResponse:
+        return await self.__command("change_profile_picture",
+                                    use_fbuploader="true", upload_id=upload_id)
+
+    async def remove_profile_picture(self) -> CurrentUserResponse:
+        return await self.__command("remove_profile_picture")
+
+    async def set_private(self, private: bool) -> CurrentUserResponse:
+        return await self.__command("set_private" if private else "set_public")
+
+    async def confirm_email(self, slug: str) -> CurrentUserResponse:
+        # slug can contain slashes, but it shouldn't start or end with one
+        return await self.__command(f"confirm_email/{slug}")
+
+    async def send_recovery_flow_email(self, query: str):
+        req = {
+            "_csrftoken": self.state.cookies.csrf_token,
+            "adid": "",
+            "guid": self.state.device.uuid,
+            "device_id": self.state.device.id,
+            "query": query,
+        }
+        resp = await self.http.post(self.url / "api/v1/accounts/send_recovery_flow_email/",
+                                    data=self.sign(req, filter_nulls=True))
+        # TODO parse response content
+        return await self.handle_response(resp)
+
+    async def edit_profile(self, external_url: Optional[str] = None, gender: Optional[str] = None,
+                           phone_number: Optional[str] = None, username: Optional[str] = None,
+                           # TODO should there be a last_name?
+                           first_name: Optional[str] = None, biography: Optional[str] = None,
+                           email: Optional[str] = None) -> CurrentUserResponse:
+        return await self.__command("edit_profile", device_id=self.state.device.id, email=email,
+                                    external_url=external_url, first_name=first_name,
+                                    username=username, phone_number=phone_number, gender=gender,
+                                    biography=biography)
+
+    async def __command(self, command: str, response_type: Type[T] = CurrentUserResponse,
+                        **kwargs: str) -> T:
+        req = {
+            "_csrftoken": self.state.cookies.csrf_token,
+            "_uid": self.state.cookies.user_id,
+            "_uuid": self.state.device.uuid,
+            **kwargs,
+        }
+        resp = await self.http.post(self.url / f"api/v1/accounts/{command}",
+                                    data=self.sign(req, filter_nulls=True))
+        return response_type.deserialize(await self.handle_response(resp))
+
+    async def read_msisdn_header(self, usage: str = "default"):
+        req = {
+            "mobile_subno_usage": usage,
+            "device_id": self.state.device.uuid,
+        }
+        headers = {
+            "X-DEVICE-ID": self.state.device.uuid,
+        }
+        resp = await self.http.post(self.url / "api/v1/accounts/read_msisdn_header/",
+                                    data=self.sign(req), headers=headers)
+        # TODO parse response content
+        return await self.handle_response(resp)
+
+    async def msisdn_header_bootstrap(self, usage: str = "default"):
+        req = {
+            "mobile_subno_usage": usage,
+            "device_id": self.state.device.uuid,
+        }
+        resp = await self.http.post(self.url / "api/v1/accounts/msisdn_header_bootstrap/",
+                                    data=self.sign(req))
+        # TODO parse response content
+        return await self.handle_response(resp)
+
+    async def contact_point_prefill(self, usage: str = "default"):
+        req = {
+            "mobile_subno_usage": usage,
+            "device_id": self.state.device.uuid,
+        }
+        resp = await self.http.post(self.url / "api/v1/accounts/contact_point_prefill/",
+                                    data=self.sign(req))
+        # TODO parse response content
+        return await self.handle_response(resp)
+
+    async def get_prefill_candidates(self):
+        req = {
+            "android_device_id": self.state.device.id,
+            "usages": json.dumps(["account_recovery_omnibox"]),
+            "device_id": self.state.device.uuid,
+        }
+        resp = await self.http.post(self.url / "api/v1/accounts/contact_point_prefill/",
+                                    data=self.sign(req))
+        # TODO parse response content
+        return await self.handle_response(resp)
+
+    async def process_contact_point_signals(self):
+        req = {
+            "phone_id": self.state.device.phone_id,
+            "_csrftoken": self.state.cookies.csrf_token,
+            "_uid": self.state.cookies.user_id,
+            "device_id": self.state.device.uuid,
+            "_uuid": self.state.device.uuid,
+            "google_tokens": json.dumps([]),
+        }
+        resp = await self.http.post(self.url / "api/v1/accounts/process_contact_point_signals/",
+                                    data=self.sign(req))
+        # TODO parse response content
+        return await self.handle_response(resp)

mauigpapi/http/api.py

@@ -1,5 +1,6 @@
 from .direct_inbox_feed import DirectInboxAPI
+from .login_simulate import LoginSimulateAPI
 
 
-class AndroidAPI(DirectInboxAPI):
+class AndroidAPI(DirectInboxAPI, LoginSimulateAPI):
     pass

mauigpapi/http/attribution.py

@@ -0,0 +1,40 @@
+# mautrix-instagram - A Matrix-Instagram puppeting bridge.
+# Copyright (C) 2020 Tulir Asokan
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Affero General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <https://www.gnu.org/licenses/>.
+from .base import BaseAndroidAPI
+from ..errors import IGResponseError
+
+
+class LogAttributionAPI(BaseAndroidAPI):
+    async def log_attribution(self):
+        resp = await self.http.get(self.url / "api/v1/attribution/log_attribution/",
+                                   data=self.sign({"adid": self.state.device.adid}))
+        # TODO parse response content
+        return await self.handle_response(resp)
+
+    async def log_resurrect_attribution(self):
+        req = {
+            "_csrftoken": self.state.cookies.csrf_token,
+            "_uid": self.state.cookies.user_id,
+            "adid": self.state.device.adid,
+            "_uuid": self.state.device.uuid,
+        }
+        resp = await self.http.get(self.url / "api/v1/attribution/log_resurrect_attribution/",
+                                   data=self.sign(req))
+        # Apparently this throws an error in the official app, so we catch it and return the error
+        try:
+            return await self.handle_response(resp)
+        except IGResponseError as e:
+            return e

mauigpapi/http/base.py

@@ -13,15 +13,21 @@
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <https://www.gnu.org/licenses/>.
-from typing import Optional, Dict
+from typing import Optional, Dict, Any
 import random
 import time
+import json
 
 from aiohttp import ClientSession, ClientResponse
 from yarl import URL
-from mautrix.types import JSON
+from mautrix.types import JSON, Serializable
 
 from ..state import AndroidState
+from ..errors import (IGActionSpamError, IGNotFoundError, IGRateLimitError, IGCheckpointError,
+                      IGUserHasLoggedOutError, IGLoginRequiredError, IGPrivateUserError,
+                      IGSentryBlockError, IGInactiveUserError, IGResponseError,
+                      IGLoginBadPasswordError, IGLoginInvalidUserError,
+                      IGLoginTwoFactorRequiredError)
 
 
 class BaseAndroidAPI:
@@ -33,12 +39,23 @@ class BaseAndroidAPI:
         self.http = ClientSession(cookie_jar=state.cookies.jar)
         self.state = state
 
+    @staticmethod
+    def sign(req: Any, filter_nulls: bool = False) -> Dict[str, str]:
+        if isinstance(req, Serializable):
+            req = req.serialize()
+        if isinstance(req, dict):
+            def remove_nulls(d: dict) -> dict:
+                return {k: v for k, v in d.items() if v is not None}
+
+            req = json.dumps(req, object_hook=remove_nulls if filter_nulls else None)
+        return {"signed_body": f"SIGNATURE.{req}"}
+
     @property
     def headers(self) -> Dict[str, str]:
         headers = {
             "User-Agent": self.state.user_agent,
             "X-Ads-Opt-Out": str(int(self.state.session.ads_opt_out)),
-            #"X-DEVICE--ID": self.state.device.uuid,
+            # "X-DEVICE--ID": self.state.device.uuid,
             "X-CM-Bandwidth-KBPS": "-1.000",
             "X-CM-Latency": "-1.000",
             "X-IG-App-Locale": self.state.device.language,
@@ -50,7 +67,7 @@ class BaseAndroidAPI:
             "X-IG-Bandwidth-TotalBytes-B": "0",
             "X-IG-Bandwidth-TotalTime-MS": "0",
             "X-IG-EU-DC-ENABLED": (str(self.state.session.eu_dc_enabled).lower()
-                                   if self.state.session.eu_dc_enabled else None),
+                                   if self.state.session.eu_dc_enabled is not None else None),
             "X-IG-Extended-CDN-Thumbnail-Cache-Busting-Value":
                 str(self.state.session.thumbnail_cache_busting_value),
             "X-Bloks-Version-Id": self.state.application.BLOKS_VERSION_ID,
@@ -79,10 +96,44 @@ class BaseAndroidAPI:
             await self._raise_response_error(resp)
 
     async def _raise_response_error(self, resp: ClientResponse) -> None:
-        # TODO handle all errors
-        print("Error:", resp.status)
-        print(await resp.json())
-        raise Exception("oh noes")
+        try:
+            data = await resp.json()
+        except json.JSONDecodeError:
+            data = {}
+
+        if data.get("spam", False):
+            raise IGActionSpamError(resp, data)
+        elif data.get("two_factor_required", False):
+            raise IGLoginTwoFactorRequiredError(resp, data)
+        elif resp.status == 404:
+            raise IGNotFoundError(resp, data)
+        elif resp.status == 429:
+            raise IGRateLimitError(resp, data)
+
+        message = data.get("message")
+        if isinstance(message, str):
+            if message == "challenge_required":
+                err = IGCheckpointError(resp, data)
+                self.state.challenge_path = err.url
+                raise err
+            elif message == "user_has_logged_out":
+                raise IGUserHasLoggedOutError(resp, data)
+            elif message == "login_required":
+                raise IGLoginRequiredError(resp, data)
+            elif message.lower() == "not authorized to view user":
+                raise IGPrivateUserError(resp, data)
+
+        error_type = data.get("error_type")
+        if error_type == "sentry_block":
+            raise IGSentryBlockError(resp, data)
+        elif error_type == "inactive_user":
+            raise IGInactiveUserError(resp, data)
+        elif error_type == "bad_password":
+            raise IGLoginBadPasswordError(resp, data)
+        elif error_type == "invalid_user":
+            raise IGLoginInvalidUserError(resp, data)
+
+        raise IGResponseError(resp, data)
 
     def _handle_response_headers(self, resp: ClientResponse) -> None:
         fields = {

mauigpapi/http/direct_inbox_feed.py

@@ -13,23 +13,10 @@
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <https://www.gnu.org/licenses/>.
-from typing import Optional, Any
-
-from attr import dataclass
-from mautrix.types import SerializableAttrs
+from typing import Optional
 
 from .base import BaseAndroidAPI
-
-
-@dataclass
-class DirectInboxResponse(SerializableAttrs['DirectInboxFeedResponse']):
-    status: str
-    seq_id: int
-    snapshot_at_ms: int
-    pending_requests_total: int
-    # TODO
-    inbox: Any
-    most_recent_inviter: Any = None
+from ..types import DirectInboxResponse
 
 
 class DirectInboxAPI(BaseAndroidAPI):

mauigpapi/http/launcher.py

@@ -0,0 +1,87 @@
+# mautrix-instagram - A Matrix-Instagram puppeting bridge.
+# Copyright (C) 2020 Tulir Asokan
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Affero General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <https://www.gnu.org/licenses/>.
+from typing import Dict, Any
+
+from .base import BaseAndroidAPI
+
+pre_login_configs = ("ig_fbns_blocked,ig_android_felix_release_players,"
+                     "ig_user_mismatch_soft_error,ig_android_carrier_signals_killswitch,"
+                     "ig_android_killswitch_perm_direct_ssim,fizz_ig_android,"
+                     "ig_mi_block_expired_events,ig_android_os_version_blocking_config")
+post_login_configs = ("ig_android_insights_welcome_dialog_tooltip,"
+                      "ig_android_extra_native_debugging_info,"
+                      "ig_android_insights_top_account_dialog_tooltip,"
+                      "ig_android_explore_startup_prefetch_launcher,"
+                      "ig_android_newsfeed_recyclerview,ig_android_react_native_ota_kill_switch,"
+                      "ig_qe_value_consistency_checker,"
+                      "ig_android_qp_keep_promotion_during_cooldown,"
+                      "ig_launcher_ig_explore_post_chaining_hide_comments_android_v0,"
+                      "ig_android_video_playback,"
+                      "ig_launcher_ig_android_network_stack_queue_undefined_request_qe,"
+                      "ig_camera_android_attributed_effects_endpoint_api_query_config,"
+                      "ig_android_notification_setting_sync,ig_android_dogfooding,"
+                      "ig_launcher_ig_explore_post_chaining_pill_android_v0,"
+                      "ig_android_request_compression_launcher,ig_delink_lasso_accounts,"
+                      "ig_android_stories_send_preloaded_reels_with_reels_tray,"
+                      "ig_android_critical_path_manager,"
+                      "ig_android_shopping_django_product_search,ig_android_qp_surveys_v1,"
+                      "ig_android_feed_attach_report_logs,ig_android_uri_parser_cache_launcher,"
+                      "ig_android_global_scheduler_infra,ig_android_explore_grid_viewpoint,"
+                      "ig_android_global_scheduler_direct,ig_android_upload_heap_on_oom,"
+                      "ig_launcher_ig_android_network_stack_cap_api_request_qe,"
+                      "ig_android_async_view_model_launcher,ig_android_bug_report_screen_record,"
+                      "ig_canvas_ad_pixel,ig_android_bloks_demos,"
+                      "ig_launcher_force_switch_on_dialog,ig_story_insights_entry,"
+                      "ig_android_executor_limit_per_group_config,"
+                      "ig_android_bitmap_strong_ref_cache_layer_launcher,"
+                      "ig_android_cold_start_class_preloading,"
+                      "ig_direct_e2e_send_waterfall_sample_rate_config,"
+                      "ig_android_qp_waterfall_logging,ig_synchronous_account_switch,"
+                      "ig_launcher_ig_android_reactnative_realtime_ota,"
+                      "ig_contact_invites_netego_killswitch,"
+                      "ig_launcher_ig_explore_video_chaining_container_module_android,"
+                      "ig_launcher_ig_explore_remove_topic_channel_tooltip_experiment_android,"
+                      "ig_android_request_cap_tuning_with_bandwidth,"
+                      "ig_android_rageshake_redesign,"
+                      "ig_launcher_explore_navigation_redesign_android,"
+                      "ig_android_betamap_cold_start,ig_android_employee_options,"
+                      "ig_android_direct_gifs_killswitch,ig_android_gps_improvements_launcher,"
+                      "ig_launcher_ig_android_network_stack_cap_video_request_qe,"
+                      "ig_launcher_ig_android_network_request_cap_tuning_qe,"
+                      "ig_android_qp_xshare_to_fb,ig_android_feed_report_ranking_issue,"
+                      "ig_launcher_ig_explore_verified_badge_android,"
+                      "ig_android_bloks_data_release,ig_android_feed_camera_latency")
+
+
+class LauncherSyncAPI(BaseAndroidAPI):
+    async def launcher_pre_login_sync(self):
+        await self.__sync({
+            "id": self.state.device.uuid,
+            "configs": pre_login_configs,
+        })
+
+    async def launcher_post_login_sync(self):
+        await self.__sync({
+            "_csrftoken": self.state.cookies.csrf_token,
+            "id": self.state.cookies.user_id,
+            "_uid": self.state.cookies.user_id,
+            "_uuid": self.state.device.uuid,
+            "configs": post_login_configs,
+        })
+
+    async def __sync(self, req: Dict[str, Any]):
+        resp = await self.http.get(self.url / "api/v1/launcher/sync/", data=self.sign(req))
+        return await self.handle_response(resp)

mauigpapi/http/login.py

@@ -0,0 +1,152 @@
+# mautrix-instagram - A Matrix-Instagram puppeting bridge.
+# Copyright (C) 2020 Tulir Asokan
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Affero General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <https://www.gnu.org/licenses/>.
+from typing import Optional
+import base64
+import struct
+import time
+import json
+import io
+
+from Crypto.PublicKey import RSA
+from Crypto.Cipher import PKCS1_OAEP, AES
+from Crypto.Random import get_random_bytes
+
+from ..types import LoginResponse, LoginResponseUser, LogoutResponse
+from .base import BaseAndroidAPI
+
+
+class LoginAPI(BaseAndroidAPI):
+    async def login(self, username: str, password: Optional[str] = None,
+                    encrypted_password: Optional[str] = None) -> LoginResponse:
+        if password:
+            if encrypted_password:
+                raise ValueError("Only one of password or encrypted_password must be provided")
+            encrypted_password = self._encrypt_password(password)
+        elif not encrypted_password:
+            raise ValueError("One of password or encrypted_password is required")
+        req = {
+            "username": username,
+            "enc_password": encrypted_password,
+            "guid": self.state.device.uuid,
+            "phone_id": self.state.device.phone_id,
+            "_csrftoken": self.state.cookies.csrf_token,
+            "device_id": self.state.device.id,
+            "adid": "",  # not set on pre-login
+            "google_tokens": "[]",
+            "login_attempt_count": 0,  # TODO maybe cache this somewhere?
+            "country_codes": json.dumps([{"country_code": "1", "source": "default"}]),
+            "jazoest": self._jazoest,
+        }
+        resp = await self.http.post(url=self.url / "api/v1/accounts/login/", data=self.sign(req))
+        return LoginResponse.deserialize(await self.handle_response(resp))
+
+    async def one_tap_app_login(self, user_id: str, nonce: str) -> LoginResponse:
+        req = {
+            "phone_id": self.state.device.phone_id,
+            "_csrftoken": self.state.cookies.csrf_token,
+            "user_id": user_id,
+            "adid": self.state.device.adid,
+            "guid": self.state.device.uuid,
+            "device_id": self.state.device.id,
+            "login_nonce": nonce,
+        }
+        resp = await self.http.post(url=self.url / "api/v1/accounts/one_tap_app_login/",
+                                    data=self.sign(req))
+        return LoginResponse.deserialize(await self.handle_response(resp))
+
+    async def two_factor_login(self, username: str, code: str, identifier: str,
+                               trust_device: bool = True, method: Optional[str] = "1"
+                               ) -> LoginResponseUser:
+        req = {
+            "verification_code": code,
+            "_csrftoken": self.state.cookies.csrf_token,
+            "two_factor_identifier": identifier,
+            "username": username,
+            "trust_this_device": "1" if trust_device else "0",
+            "guid": self.state.device.uuid,
+            "device_id": self.state.device.id,
+            "verification_method": method,
+        }
+        resp = await self.http.post(url=self.url / "api/v1/accounts/one_tap_app_login/",
+                                    data=self.sign(req))
+        return LoginResponseUser.deserialize(await self.handle_response(resp))
+
+    async def logout(self, one_tap_app_login: Optional[bool] = None) -> LogoutResponse:
+        req = {
+            "guid": self.state.device.uuid,
+            "phone_id": self.state.device.phone_id,
+            "_csrftoken": self.state.cookies.csrf_token,
+            "device_id": self.state.device.id,
+            "_uuid": self.state.device.uuid,
+            "one_tap_app_login": one_tap_app_login,
+        }
+        resp = await self.http.post(url=self.url / "api/v1/accounts/logout/",
+                                    data=self.sign(req))
+        return LogoutResponse.deserialize(await self.handle_response(resp))
+
+    async def change_password(self, old_password: str, new_password: str):
+        return self.change_password_encrypted(old_password=self._encrypt_password(old_password),
+                                              new_password1=self._encrypt_password(new_password),
+                                              new_password2=self._encrypt_password(new_password))
+
+    async def change_password_encrypted(self, old_password: str, new_password1: str,
+                                        new_password2: str):
+        req = {
+            "_csrftoken": self.state.cookies.csrf_token,
+            "_uid": self.state.cookies.user_id,
+            "_uuid": self.state.device.uuid,
+            "enc_old_password": old_password,
+            "enc_new_password1": new_password1,
+            "enc_new_password2": new_password2,
+        }
+        resp = await self.http.post(self.url / "api/v1/accounts/change_password/",
+                                    data=self.sign(req))
+        # TODO parse response content
+        return await self.handle_response(resp)
+
+    def _encrypt_password(self, password: str) -> str:
+        # Key and IV for AES encryption
+        rand_key = get_random_bytes(32)
+        iv = get_random_bytes(12)
+
+        # Encrypt AES key with Instagram's RSA public key
+        pubkey_bytes = base64.b64decode(self.state.session.password_encryption_pubkey)
+        pubkey = RSA.import_key(pubkey_bytes)
+        cipher_rsa = PKCS1_OAEP.new(pubkey)
+        encrypted_rand_key = cipher_rsa.encrypt(rand_key)
+
+        cipher_aes = AES.new(rand_key, AES.MODE_GCM, iv=iv)
+        # Add the current time to the additional authenticated data (AAD) section
+        current_time = int(time.time())
+        cipher_aes.update(str(current_time).encode("utf-8"))
+        # Encrypt the password and get the AES MAC auth tag
+        encrypted_passwd, auth_tag = cipher_aes.encrypt_and_digest(password.encode("utf-8"))
+
+        buf = io.BytesIO()
+        # 1 is presumably the version
+        buf.write(bytes([1, int(self.state.session.password_encryption_key_id)]))
+        buf.write(iv)
+        # Length of the encrypted AES key as a little-endian 16-bit int
+        buf.write(struct.pack("<h", len(encrypted_rand_key)))
+        buf.write(encrypted_rand_key)
+        buf.write(auth_tag)
+        buf.write(encrypted_passwd)
+        encoded = base64.b64encode(buf.getvalue())
+        return f"#PWD_INSTAGRAM:4:{current_time}:{encoded}"
+
+    @property
+    def _jazoest(self) -> str:
+        return f"2{sum(ord(i) for i in self.state.device.phone_id)}"

mauigpapi/http/login_simulate.py

@@ -0,0 +1,83 @@
+# mautrix-instagram - A Matrix-Instagram puppeting bridge.
+# Copyright (C) 2020 Tulir Asokan
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Affero General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <https://www.gnu.org/licenses/>.
+from typing import List, Awaitable, Any
+import random
+
+from ..types import LoginResponse
+from .account import AccountAPI
+from .login import LoginAPI
+from .qe import QeSyncAPI
+from .zr import ZRTokenAPI
+from .attribution import LogAttributionAPI
+from .launcher import LauncherSyncAPI
+
+
+class LoginSimulateAPI(AccountAPI, LogAttributionAPI, QeSyncAPI, ZRTokenAPI, LoginAPI,
+                       LauncherSyncAPI):
+    @property
+    def _pre_login_flow_requests(self) -> List[Awaitable[Any]]:
+        return [
+            self.read_msisdn_header(),
+            self.msisdn_header_bootstrap("ig_select_app"),
+            self.zr_token_result(),
+            self.contact_point_prefill("prefill"),
+            self.launcher_pre_login_sync(),
+            self.qe_sync_login_experiments(),
+            self.log_attribution(),
+            self.get_prefill_candidates(),
+        ]
+
+    @property
+    def _post_login_flow_requests(self) -> List[Awaitable[Any]]:
+        return [
+            self.zr_token_result(),
+            self.launcher_post_login_sync(),
+            self.qe_sync_experiments(),
+            self.log_attribution(),
+            self.log_resurrect_attribution(),
+
+            self._facebook_ota(),
+        ]
+
+    async def simulate_pre_login_flow(self) -> None:
+        items = self._pre_login_flow_requests
+        random.shuffle(items)
+        for item in items:
+            await item
+
+    async def simulate_post_login_flow(self) -> None:
+        items = self._post_login_flow_requests
+        random.shuffle(items)
+        for item in items:
+            await item
+
+    async def _facebook_ota(self):
+        url = (self.url / "api/v1/facebook_ota/").with_query({
+            "fields": self.state.application.FACEBOOK_OTA_FIELDS,
+            "custom_user_id": self.state.cookies.user_id,
+            "signed_body": "SIGNATURE.",
+            "version_code": self.state.application.APP_VERSION_CODE,
+            "version_name": self.state.application.APP_VERSION,
+            "custom_app_id": self.state.application.FACEBOOK_ORCA_APPLICATION_ID,
+            "custom_device_id": self.state.device.uuid,
+        })
+        resp = await self.http.get(url)
+        return await self.handle_response(resp)
+
+    async def upgrade_login(self) -> LoginResponse:
+        user_id = self.state.cookies.user_id
+        resp = await self.logout(one_tap_app_login=True)
+        return await self.one_tap_app_login(user_id, resp.login_nonce)

mauigpapi/http/qe.py

@@ -0,0 +1,28 @@
+# mautrix-instagram - A Matrix-Instagram puppeting bridge.
+# Copyright (C) 2020 Tulir Asokan
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Affero General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <https://www.gnu.org/licenses/>.
+from .base import BaseAndroidAPI
+
+
+class QeSyncAPI(BaseAndroidAPI):
+    async def qe_sync_experiments(self) -> None:
+        await self.__sync(self.state.application.EXPERIMENTS)
+
+    async def qe_sync_login_experiments(self) -> None:
+        await self.__sync(self.state.application.LOGIN_EXPERIMENTS)
+
+    async def __sync(self, experiments: str) -> None:
+        # TODO implement
+        pass

mauigpapi/http/zr.py

@@ -0,0 +1,29 @@
+# mautrix-instagram - A Matrix-Instagram puppeting bridge.
+# Copyright (C) 2020 Tulir Asokan
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Affero General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <https://www.gnu.org/licenses/>.
+from .base import BaseAndroidAPI
+
+
+class ZRTokenAPI(BaseAndroidAPI):
+    async def zr_token_result(self):
+        url = (self.url / "api/v1/zr/token/result/").with_query({
+            "device_id": self.state.device.id,
+            "token_hash": "",
+            "custom_device_id": self.state.device.uuid,
+            "fetch_reason": "token_expired",
+        })
+        resp = await self.http.get(url)
+        # TODO parse response content
+        return await self.handle_response(resp)

mauigpapi/mqtt/__init__.py

@@ -1,4 +1,16 @@
 from .subscription import SkywalkerSubscription, GraphQLSubscription
 from .types import (RealtimeTopic, ThreadItemType, ThreadAction, ReactionStatus, TypingStatus,
-                    CommandResponse, CommandResponsePayload)
+                    CommandResponse, CommandResponsePayload, Operation, IrisPayload, ImageVersions,
+                    IrisPayloadData, ViewMode, CreativeConfig, CreateModeAttribution, ImageVersion,
+                    VideoVersion, MediaType,RegularMediaItem, FriendshipStatus, MinimalUser, User,
+                    Caption, MediaShareItem, ReplayableMediaItem, VisualMedia, AudioInfo,
+                    VoiceMediaData, VoiceMediaItem, AnimatedMediaItem, AnimatedMediaImage,
+                    AnimatedMediaImages, MessageSyncEvent, MessageSyncMessage, PubsubPayloadData,
+                    PubsubBasePayload, PubsubPublishMetadata, PubsubPayload, PubsubEvent,
+                    ActivityIndicatorData, AppPresenceEventPayload, AppPresenceEvent,
+                    RealtimeZeroProvisionPayload, ZeroProductProvisioningEvent, RealtimeDirectData,
+                    RealtimeDirectEvent, ClientConfigUpdatePayload, ClientConfigUpdateEvent,
+                    LiveVideoCommentPayload, LiveVideoCommentUser, LiveVideoCommentEvent,
+                    LiveVideoComment, LiveVideoSystemComment)
+from .events import Connect, Disconnect
 from .conn import AndroidMQTT

mauigpapi/state/experiments.py

@@ -0,0 +1,37 @@
+# mautrix-instagram - A Matrix-Instagram puppeting bridge.
+# Copyright (C) 2020 Tulir Asokan
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Affero General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <https://www.gnu.org/licenses/>.
+from typing import Dict
+
+from mautrix.types import Serializable, JSON
+from mautrix.types.util import no_value
+from ..types import AndroidExperiment, QeSyncResponse
+
+
+class AndroidExperiments(Serializable):
+    experiments: Dict[str, AndroidExperiment]
+
+    def __init__(self) -> None:
+        self.experiments = {}
+
+    def update(self, updated: QeSyncResponse) -> None:
+        self.experiments.update({item.name: item.parse() for item in updated.experiments})
+
+    def serialize(self) -> no_value:
+        return no_value
+
+    @classmethod
+    def deserialize(cls, raw: JSON) -> 'AndroidExperiments':
+        return cls()

mauigpapi/state/state.py

@@ -27,6 +27,7 @@ from ..errors import IGNoCheckpointError, IGCookieNotFoundError, IGUserIDNotFoun
 from .device import AndroidDevice
 from .session import AndroidSession
 from .application import AndroidApplication
+from .experiments import AndroidExperiments
 from .cookies import Cookies
 
 
@@ -35,7 +36,7 @@ class AndroidState(SerializableAttrs['AndroidState']):
     device: AndroidDevice = attr.ib(factory=lambda: AndroidDevice())
     session: AndroidSession = attr.ib(factory=lambda: AndroidSession())
     application: AndroidApplication = attr.ib(factory=lambda: AndroidApplication())
-    # experiments: AndroidExperiments
+    experiments: AndroidExperiments = attr.ib(factory=lambda: AndroidExperiments())
     client_session_id_lifetime: int = 1_200_000
     pigeon_session_id_lifetime: int = 1_200_000
     challenge: 'Optional[ChallengeStateResponse]' = None
@@ -70,6 +71,10 @@ class AndroidState(SerializableAttrs['AndroidState']):
             raise IGNoCheckpointError()
         return self._challenge_path
 
+    @challenge_path.setter
+    def challenge_path(self, val: str) -> None:
+        self._challenge_path = val
+
     def _gen_temp_uuid(self, seed: str, lifetime: int) -> UUID:
         rand = random.Random(f"{seed}{self.device.id}{round(time.time() * 1000 / lifetime)}")
         return UUID(int=rand.getrandbits(128), version=4)

mauigpapi/types/__init__.py

@@ -0,0 +1,7 @@
+from .qe import AndroidExperiment, QeSyncExperiment, QeSyncExperimentParam, QeSyncResponse
+from .error import (SpamResponse, CheckpointResponse, CheckpointChallenge,
+                    LoginRequiredResponse, LoginErrorResponse, LoginErrorResponseButton,
+                    LoginPhoneVerificationSettings, LoginTwoFactorInfo)
+from .login import LoginResponseUser, LoginResponseNametag, LoginResponse, LogoutResponse
+from .account import CurrentUser, EntityText, HDProfilePictureVersion, CurrentUserResponse
+from .direct_inbox import DirectInboxResponse

mauigpapi/types/account.py

@@ -0,0 +1,77 @@
+# mautrix-instagram - A Matrix-Instagram puppeting bridge.
+# Copyright (C) 2020 Tulir Asokan
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Affero General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <https://www.gnu.org/licenses/>.
+from typing import Any, List
+
+from attr import dataclass
+
+from mautrix.types import SerializableAttrs
+
+
+@dataclass
+class BaseResponseUser(SerializableAttrs['BaseResponseUser']):
+    pk: int
+    username: str
+    full_name: str
+    is_private: bool
+    profile_pic_url: str
+    profile_pic_id: str
+    is_verified: bool
+    has_anonymous_profile_picture: bool
+
+    phone_number: str
+    country_code: int
+    national_number: int
+
+    reel_auto_archive: str
+    allowed_commenter_type: str
+
+
+@dataclass
+class EntityText(SerializableAttrs['EntityText']):
+    raw_text: str
+    # TODO figure out type
+    entities: List[Any]
+
+
+@dataclass
+class HDProfilePictureVersion(SerializableAttrs['HDProfilePictureVersion']):
+    url: str
+    width: int
+    height: int
+
+
+# Not sure if these are actually the same
+HDProfilePictureURLInfo = HDProfilePictureVersion
+
+
+@dataclass
+class CurrentUser(SerializableAttrs['CurrentUser']):
+    biography: str
+    can_link_entities_in_bio: bool
+    biography_with_entities: EntityText
+    external_url: str
+    has_biography_translation: bool
+    hd_profile_pic_versions: HDProfilePictureVersion
+    hd_profile_pic_url_info: HDProfilePictureURLInfo
+    show_conversion_edit_entry: bool
+    birthday: Any
+    gender: int
+    email: str
+
+@dataclass
+class CurrentUserResponse(SerializableAttrs['CurrentUserResponse']):
+    status: str
+    user: CurrentUser

mauigpapi/types/direct_inbox.py

@@ -0,0 +1,30 @@
+# mautrix-instagram - A Matrix-Instagram puppeting bridge.
+# Copyright (C) 2020 Tulir Asokan
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Affero General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <https://www.gnu.org/licenses/>.
+from typing import Any
+
+from attr import dataclass
+from mautrix.types import SerializableAttrs
+
+
+@dataclass
+class DirectInboxResponse(SerializableAttrs['DirectInboxFeedResponse']):
+    status: str
+    seq_id: int
+    snapshot_at_ms: int
+    pending_requests_total: int
+    # TODO
+    inbox: Any
+    most_recent_inviter: Any = None

mauigpapi/types/error.py

@@ -0,0 +1,100 @@
+# mautrix-instagram - A Matrix-Instagram puppeting bridge.
+# Copyright (C) 2020 Tulir Asokan
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Affero General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <https://www.gnu.org/licenses/>.
+from typing import Optional, List
+
+from attr import dataclass
+
+from mautrix.types import SerializableAttrs
+
+
+@dataclass
+class SpamResponse(SerializableAttrs['SpamResponse']):
+    feedback_title: str
+    feedback_message: str
+    feedback_url: str
+    feedback_appeal_label: str
+    feedback_ignore_label: str
+    feedback_action: str
+    message: str = "feedback_required"
+    spam: bool = True
+    status: str = "fail"
+    error_type: Optional[str] = None
+
+
+@dataclass
+class CheckpointChallenge(SerializableAttrs['CheckpointChallenge']):
+    url: str
+    api_path: str
+    hide_webview_header: bool
+    lock: bool
+    logout: bool
+    native_flow: bool
+
+
+@dataclass
+class CheckpointResponse(SerializableAttrs['CheckpointResponse']):
+    message: str  # challenge_required
+    status: str  # fail
+    error_type: str
+    challenge: CheckpointChallenge
+
+
+@dataclass
+class LoginRequiredResponse(SerializableAttrs['LoginRequiredResponse']):
+    # TODO enum?
+    logout_reason: int
+    message: str  # login_required or user_has_logged_out
+    status: str  # fail
+
+
+@dataclass
+class LoginErrorResponseButton(SerializableAttrs['LoginErrorResponseButton']):
+    title: str
+    action: str
+
+
+@dataclass
+class LoginPhoneVerificationSettings(SerializableAttrs['LoginPhoneVerificationSettings']):
+    max_sms_count: int
+    resend_sms_delay_sec: int
+    robocall_count_down_time_sec: int
+    robocall_max_after_sms: bool
+
+
+@dataclass
+class LoginTwoFactorInfo(SerializableAttrs['LoginTwoFactorInfo']):
+    username: str
+    sms_two_factor_on: bool
+    totp_two_factor_on: bool
+    obfuscated_phone_number: str
+    two_factor_identifier: str
+    show_messenger_code_option: bool
+    show_new_login_screen: bool
+    show_trusted_device_option: bool
+    phone_verification_settings: Optional[LoginPhoneVerificationSettings] = None
+
+
+@dataclass
+class LoginErrorResponse(SerializableAttrs['LoginErrorResponse']):
+    message: str
+    status: str
+    error_type: str
+    error_title: str
+    buttons: List[LoginErrorResponseButton]
+    invalid_credentials: bool
+    two_factor_required: bool
+    two_factor_info: Optional[LoginTwoFactorInfo] = None
+    phone_verification_settings: Optional[LoginPhoneVerificationSettings] = None

mauigpapi/types/login.py

@@ -0,0 +1,55 @@
+# mautrix-instagram - A Matrix-Instagram puppeting bridge.
+# Copyright (C) 2020 Tulir Asokan
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Affero General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <https://www.gnu.org/licenses/>.
+from typing import Any, Optional
+
+from attr import dataclass
+
+from mautrix.types import SerializableAttrs
+
+from .account import BaseResponseUser
+
+
+@dataclass
+class LoginResponseNametag(SerializableAttrs['LoginResponseNametag']):
+    mode: int
+    gradient: str
+    emoji: str
+    selfie_sticker: str
+
+
+@dataclass
+class LoginResponseUser(BaseResponseUser, SerializableAttrs['LoginResponseUser']):
+    can_boost_post: bool
+    is_business: bool
+    account_type: int
+    is_call_to_action_enabled: Any
+    can_see_organic_insights: bool
+    show_insights_terms: bool
+    has_placed_orders: bool
+    nametag: LoginResponseNametag
+    allow_contacts_sync: bool
+
+
+@dataclass
+class LoginResponse(SerializableAttrs['LoginResponse']):
+    logged_in_user: LoginResponseUser
+    status: str
+
+
+@dataclass
+class LogoutResponse(SerializableAttrs['LogoutResponse']):
+    status: str
+    login_nonce: Optional[str] = None

mauigpapi/types/qe.py

@@ -0,0 +1,63 @@
+# mautrix-instagram - A Matrix-Instagram puppeting bridge.
+# Copyright (C) 2020 Tulir Asokan
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Affero General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <https://www.gnu.org/licenses/>.
+from typing import List, Optional, Any, Dict
+import json
+
+from attr import dataclass, attrib
+
+from mautrix.types import SerializableAttrs
+
+
+@dataclass
+class QeSyncExperimentParam(SerializableAttrs['QeSyncExperimentParam']):
+    name: str
+    value: str
+
+
+@dataclass
+class AndroidExperiment:
+    group: str
+    params: Dict[str, Any] = attrib(factory=lambda: {})
+    additional: List[Any] = attrib(factory=lambda: [])
+    logging_id: Optional[str] = None
+
+
+def _try_parse(val: str) -> Any:
+    try:
+        return json.loads(val)
+    except json.JSONDecodeError:
+        return val
+
+
+@dataclass
+class QeSyncExperiment(SerializableAttrs['QeSyncResponseExperiment']):
+    name: str
+    group: str
+    additional_params: List[Any]
+    params: List[QeSyncExperimentParam]
+    logging_id: Optional[str] = None
+
+    def parse(self) -> AndroidExperiment:
+        return AndroidExperiment(group=self.group, additional=self.additional_params,
+                                 logging_id=self.logging_id,
+                                 params={param.name: _try_parse(param.value)
+                                         for param in self.params})
+
+
+@dataclass
+class QeSyncResponse(SerializableAttrs['QeSyncResponse']):
+    experiments: List[QeSyncExperiment]
+    status: str