login.py 7.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180
  1. # mautrix-instagram - A Matrix-Instagram puppeting bridge.
  2. # Copyright (C) 2022 Tulir Asokan
  3. #
  4. # This program is free software: you can redistribute it and/or modify
  5. # it under the terms of the GNU Affero General Public License as published by
  6. # the Free Software Foundation, either version 3 of the License, or
  7. # (at your option) any later version.
  8. #
  9. # This program is distributed in the hope that it will be useful,
  10. # but WITHOUT ANY WARRANTY; without even the implied warranty of
  11. # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
  12. # GNU Affero General Public License for more details.
  13. #
  14. # You should have received a copy of the GNU Affero General Public License
  15. # along with this program. If not, see <https://www.gnu.org/licenses/>.
  16. from __future__ import annotations
  17. import base64
  18. import io
  19. import json
  20. import struct
  21. import time
  22. from Crypto.Cipher import AES, PKCS1_v1_5
  23. from Crypto.PublicKey import RSA
  24. from Crypto.Random import get_random_bytes
  25. from ..types import FacebookLoginResponse, LoginResponse, LogoutResponse
  26. from .base import BaseAndroidAPI
  27. class LoginAPI(BaseAndroidAPI):
  28. async def login(
  29. self,
  30. username: str,
  31. password: str | None = None,
  32. encrypted_password: str | None = None,
  33. ) -> LoginResponse:
  34. if password:
  35. if encrypted_password:
  36. raise ValueError("Only one of password or encrypted_password must be provided")
  37. encrypted_password = self._encrypt_password(password)
  38. elif not encrypted_password:
  39. raise ValueError("One of password or encrypted_password is required")
  40. req = {
  41. "username": username,
  42. "enc_password": encrypted_password,
  43. "guid": self.state.device.uuid,
  44. "phone_id": self.state.device.phone_id,
  45. "device_id": self.state.device.id,
  46. "adid": self.state.device.adid,
  47. "google_tokens": "[]",
  48. "login_attempt_count": "0", # TODO maybe cache this somewhere?
  49. "country_codes": json.dumps([{"country_code": "1", "source": "default"}]),
  50. "jazoest": self._jazoest,
  51. }
  52. return await self.std_http_post(
  53. "/api/v1/accounts/login/", data=req, response_type=LoginResponse
  54. )
  55. async def one_tap_app_login(self, user_id: str, nonce: str) -> LoginResponse:
  56. req = {
  57. "phone_id": self.state.device.phone_id,
  58. "user_id": user_id,
  59. "adid": self.state.device.adid,
  60. "guid": self.state.device.uuid,
  61. "device_id": self.state.device.id,
  62. "login_nonce": nonce,
  63. }
  64. return await self.std_http_post(
  65. "/api/v1/accounts/one_tap_app_login/", data=req, response_type=LoginResponse
  66. )
  67. async def two_factor_login(
  68. self,
  69. username: str,
  70. code: str,
  71. identifier: str,
  72. trust_device: bool = True,
  73. is_totp: bool = True,
  74. ) -> LoginResponse:
  75. req = {
  76. "verification_code": code,
  77. "two_factor_identifier": identifier,
  78. "username": username,
  79. "trust_this_device": "1" if trust_device else "0",
  80. "guid": self.state.device.uuid,
  81. "device_id": self.state.device.id,
  82. "verification_method": "3" if is_totp else "1",
  83. }
  84. return await self.std_http_post(
  85. "/api/v1/accounts/two_factor_login/", data=req, response_type=LoginResponse
  86. )
  87. # async def two_factor_trusted_status(self, username: str, identifier: str, polling_nonce: str):
  88. # pass
  89. async def facebook_signup(self, fb_access_token: str) -> FacebookLoginResponse:
  90. req = {
  91. "jazoest": self._jazoest,
  92. "dryrun": "true",
  93. "fb_req_flag": "false",
  94. "phone_id": self.state.device.phone_id,
  95. "force_signup_with_fb_after_cp_claiming": "false",
  96. "adid": self.state.device.adid,
  97. "guid": self.state.device.uuid,
  98. "device_id": self.state.device.id,
  99. # "waterfall_id": uuid4(),
  100. "fb_access_token": fb_access_token,
  101. }
  102. return await self.std_http_post(
  103. "/api/v1/fb/facebook_signup/", data=req, response_type=FacebookLoginResponse
  104. )
  105. async def logout(self, one_tap_app_login: bool | None = None) -> LogoutResponse:
  106. req = {
  107. "guid": self.state.device.uuid,
  108. "phone_id": self.state.device.phone_id,
  109. "device_id": self.state.device.id,
  110. "_uuid": self.state.device.uuid,
  111. "one_tap_app_login": one_tap_app_login,
  112. }
  113. return await self.std_http_post(
  114. "/api/v1/accounts/logout/", data=req, response_type=LogoutResponse
  115. )
  116. async def change_password(self, old_password: str, new_password: str):
  117. return self.change_password_encrypted(
  118. old_password=self._encrypt_password(old_password),
  119. new_password1=self._encrypt_password(new_password),
  120. new_password2=self._encrypt_password(new_password),
  121. )
  122. async def change_password_encrypted(
  123. self, old_password: str, new_password1: str, new_password2: str
  124. ):
  125. req = {
  126. "_csrftoken": self.state.cookies.csrf_token,
  127. "_uid": self.state.session.ds_user_id,
  128. "_uuid": self.state.device.uuid,
  129. "enc_old_password": old_password,
  130. "enc_new_password1": new_password1,
  131. "enc_new_password2": new_password2,
  132. }
  133. # TODO parse response content
  134. return await self.std_http_post("/api/v1/accounts/change_password/", data=req)
  135. def _encrypt_password(self, password: str) -> str:
  136. # Key and IV for AES encryption
  137. rand_key = get_random_bytes(32)
  138. iv = get_random_bytes(12)
  139. # Encrypt AES key with Instagram's RSA public key
  140. pubkey_bytes = base64.b64decode(self.state.session.password_encryption_pubkey)
  141. pubkey = RSA.import_key(pubkey_bytes)
  142. cipher_rsa = PKCS1_v1_5.new(pubkey)
  143. encrypted_rand_key = cipher_rsa.encrypt(rand_key)
  144. cipher_aes = AES.new(rand_key, AES.MODE_GCM, nonce=iv)
  145. # Add the current time to the additional authenticated data (AAD) section
  146. current_time = int(time.time())
  147. cipher_aes.update(str(current_time).encode("utf-8"))
  148. # Encrypt the password and get the AES MAC auth tag
  149. encrypted_passwd, auth_tag = cipher_aes.encrypt_and_digest(password.encode("utf-8"))
  150. buf = io.BytesIO()
  151. # 1 is presumably the version
  152. buf.write(bytes([1, int(self.state.session.password_encryption_key_id)]))
  153. buf.write(iv)
  154. # Length of the encrypted AES key as a little-endian 16-bit int
  155. buf.write(struct.pack("<h", len(encrypted_rand_key)))
  156. buf.write(encrypted_rand_key)
  157. buf.write(auth_tag)
  158. buf.write(encrypted_passwd)
  159. encoded = base64.b64encode(buf.getvalue()).decode("utf-8")
  160. return f"#PWD_INSTAGRAM:4:{current_time}:{encoded}"
  161. @property
  162. def _jazoest(self) -> str:
  163. return f"2{sum(ord(i) for i in self.state.device.phone_id)}"