conn.py 37 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945
  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. from typing import Any, Awaitable, Callable, Iterable, Type, TypeVar
  18. from collections import defaultdict
  19. from socket import error as SocketError, socket
  20. import asyncio
  21. import json
  22. import logging
  23. import re
  24. import time
  25. import urllib.request
  26. import zlib
  27. from yarl import URL
  28. import paho.mqtt.client as pmc
  29. from mautrix.util.logging import TraceLogger
  30. from ..errors import (
  31. IrisSubscribeError,
  32. MQTTConnectionUnauthorized,
  33. MQTTNotConnected,
  34. MQTTNotLoggedIn,
  35. )
  36. from ..proxy import ProxyHandler
  37. from ..state import AndroidState
  38. from ..types import (
  39. AppPresenceEventPayload,
  40. ClientConfigUpdatePayload,
  41. CommandResponse,
  42. IrisPayload,
  43. IrisPayloadData,
  44. LiveVideoCommentPayload,
  45. MessageSyncEvent,
  46. MessageSyncMessage,
  47. Operation,
  48. PubsubEvent,
  49. PubsubPayload,
  50. ReactionStatus,
  51. ReactionType,
  52. RealtimeDirectEvent,
  53. RealtimeZeroProvisionPayload,
  54. ThreadAction,
  55. ThreadItemType,
  56. ThreadSyncEvent,
  57. TypingStatus,
  58. )
  59. from .events import Connect, Disconnect, NewSequenceID, ProxyUpdate
  60. from .otclient import MQTToTClient
  61. from .subscription import GraphQLQueryID, RealtimeTopic, everclear_subscriptions
  62. from .thrift import ForegroundStateConfig, IncomingMessage, RealtimeClientInfo, RealtimeConfig
  63. try:
  64. import socks
  65. except ImportError:
  66. socks = None
  67. T = TypeVar("T")
  68. ACTIVITY_INDICATOR_REGEX = re.compile(
  69. r"/direct_v2/threads/([\w_]+)/activity_indicator_id/([\w_]+)"
  70. )
  71. INBOX_THREAD_REGEX = re.compile(r"/direct_v2/inbox/threads/([\w_]+)")
  72. class AndroidMQTT:
  73. _loop: asyncio.AbstractEventLoop
  74. _client: MQTToTClient
  75. log: TraceLogger
  76. state: AndroidState
  77. _graphql_subs: set[str]
  78. _skywalker_subs: set[str]
  79. _iris_seq_id: int | None
  80. _iris_snapshot_at_ms: int | None
  81. _publish_waiters: dict[int, asyncio.Future]
  82. _response_waiters: dict[RealtimeTopic, asyncio.Future]
  83. _response_waiter_locks: dict[RealtimeTopic, asyncio.Lock]
  84. _message_response_waiter_lock: asyncio.Lock
  85. _message_response_waiter_id: str | None
  86. _message_response_waiter: asyncio.Future | None
  87. _disconnect_error: Exception | None
  88. _event_handlers: dict[Type[T], list[Callable[[T], Awaitable[None]]]]
  89. _outgoing_events: asyncio.Queue
  90. _event_dispatcher_task: asyncio.Task | None
  91. # region Initialization
  92. def __init__(
  93. self,
  94. state: AndroidState,
  95. loop: asyncio.AbstractEventLoop | None = None,
  96. log: TraceLogger | None = None,
  97. proxy_handler: ProxyHandler | None = None,
  98. ) -> None:
  99. self._graphql_subs = set()
  100. self._skywalker_subs = set()
  101. self._iris_seq_id = None
  102. self._iris_snapshot_at_ms = None
  103. self._publish_waiters = {}
  104. self._response_waiters = {}
  105. self._message_response_waiter_lock = asyncio.Lock()
  106. self._message_response_waiter_id = None
  107. self._message_response_waiter = None
  108. self._disconnect_error = None
  109. self._response_waiter_locks = defaultdict(lambda: asyncio.Lock())
  110. self._event_handlers = defaultdict(lambda: [])
  111. self._event_dispatcher_task = None
  112. self._outgoing_events = asyncio.Queue()
  113. self.log = log or logging.getLogger("mauigpapi.mqtt")
  114. self._loop = loop or asyncio.get_event_loop()
  115. self.state = state
  116. self._client = MQTToTClient(
  117. client_id=self._form_client_id(),
  118. clean_session=True,
  119. protocol=pmc.MQTTv31,
  120. transport="tcp",
  121. )
  122. self.proxy_handler = proxy_handler
  123. self.setup_proxy()
  124. self._client.enable_logger()
  125. self._client.tls_set()
  126. # mqtt.max_inflight_messages_set(20) # The rest will get queued
  127. # mqtt.max_queued_messages_set(0) # Unlimited messages can be queued
  128. # mqtt.message_retry_set(20) # Retry sending for at least 20 seconds
  129. # mqtt.reconnect_delay_set(min_delay=1, max_delay=120)
  130. self._client.connect_async("edge-mqtt.facebook.com", 443, keepalive=60)
  131. self._client.on_message = self._on_message_handler
  132. self._client.on_publish = self._on_publish_handler
  133. self._client.on_connect = self._on_connect_handler
  134. self._client.on_disconnect = self._on_disconnect_handler
  135. self._client.on_socket_open = self._on_socket_open
  136. self._client.on_socket_close = self._on_socket_close
  137. self._client.on_socket_register_write = self._on_socket_register_write
  138. self._client.on_socket_unregister_write = self._on_socket_unregister_write
  139. def setup_proxy(self):
  140. http_proxy = self.proxy_handler.get_proxy_url()
  141. if http_proxy:
  142. if not socks:
  143. self.log.warning("http_proxy is set, but pysocks is not installed")
  144. else:
  145. proxy_url = URL(http_proxy)
  146. proxy_type = {
  147. "http": socks.HTTP,
  148. "https": socks.HTTP,
  149. "socks": socks.SOCKS5,
  150. "socks5": socks.SOCKS5,
  151. "socks4": socks.SOCKS4,
  152. }[proxy_url.scheme]
  153. self._client.proxy_set(
  154. proxy_type=proxy_type,
  155. proxy_addr=proxy_url.host,
  156. proxy_port=proxy_url.port,
  157. proxy_username=proxy_url.user,
  158. proxy_password=proxy_url.password,
  159. )
  160. def _clear_response_waiters(self) -> None:
  161. for waiter in self._response_waiters.values():
  162. if not waiter.done():
  163. waiter.set_exception(
  164. MQTTNotConnected("MQTT disconnected before request returned response")
  165. )
  166. for waiter in self._publish_waiters.values():
  167. if not waiter.done():
  168. waiter.set_exception(
  169. MQTTNotConnected("MQTT disconnected before request was published")
  170. )
  171. self._response_waiters = {}
  172. self._publish_waiters = {}
  173. def _form_client_id(self) -> bytes:
  174. subscribe_topics = [
  175. RealtimeTopic.PUBSUB, # 88
  176. RealtimeTopic.SUB_IRIS_RESPONSE, # 135
  177. RealtimeTopic.RS_REQ, # 244
  178. RealtimeTopic.REALTIME_SUB, # 149
  179. RealtimeTopic.REGION_HINT, # 150
  180. RealtimeTopic.RS_RESP, # 245
  181. RealtimeTopic.T_RTC_LOG, # 274
  182. RealtimeTopic.SEND_MESSAGE_RESPONSE, # 133
  183. RealtimeTopic.MESSAGE_SYNC, # 146
  184. RealtimeTopic.LIGHTSPEED_RESPONSE, # 179
  185. RealtimeTopic.UNKNOWN_PP, # 34
  186. ]
  187. subscribe_topic_ids = [int(topic.encoded) for topic in subscribe_topics]
  188. password = f"authorization={self.state.session.authorization}"
  189. # if not self.state.session.authorization:
  190. # password = f"sessionid={self.state.cookies['sessionid']}"
  191. cfg = RealtimeConfig(
  192. client_identifier=self.state.device.phone_id[:20],
  193. client_info=RealtimeClientInfo(
  194. user_id=int(self.state.user_id),
  195. user_agent=self.state.user_agent,
  196. client_capabilities=0b10110111,
  197. endpoint_capabilities=0,
  198. publish_format=1,
  199. no_automatic_foreground=True,
  200. make_user_available_in_foreground=False,
  201. device_id=self.state.device.phone_id,
  202. is_initially_foreground=False,
  203. network_type=1,
  204. network_subtype=0,
  205. client_mqtt_session_id=int(time.time() * 1000) & 0xFFFFFFFF,
  206. subscribe_topics=subscribe_topic_ids,
  207. client_type="cookie_auth",
  208. app_id=567067343352427,
  209. region_preference=self.state.session.region_hint or "LLA",
  210. device_secret="",
  211. client_stack=3,
  212. ),
  213. password=password,
  214. app_specific_info={
  215. "app_version": self.state.application.APP_VERSION,
  216. "X-IG-Capabilities": self.state.application.CAPABILITIES,
  217. "everclear_subscriptions": json.dumps(everclear_subscriptions),
  218. "User-Agent": self.state.user_agent,
  219. "Accept-Language": self.state.device.language.replace("_", "-"),
  220. "platform": "android",
  221. "ig_mqtt_route": "django",
  222. "pubsub_msg_type_blacklist": "direct, typing_type",
  223. "auth_cache_enabled": "1",
  224. },
  225. )
  226. return zlib.compress(cfg.to_thrift(), level=9)
  227. # endregion
  228. def _on_socket_open(self, client: MQTToTClient, _: Any, sock: socket) -> None:
  229. self._loop.add_reader(sock, client.loop_read)
  230. def _on_socket_close(self, client: MQTToTClient, _: Any, sock: socket) -> None:
  231. self._loop.remove_reader(sock)
  232. def _on_socket_register_write(self, client: MQTToTClient, _: Any, sock: socket) -> None:
  233. self._loop.add_writer(sock, client.loop_write)
  234. def _on_socket_unregister_write(self, client: MQTToTClient, _: Any, sock: socket) -> None:
  235. self._loop.remove_writer(sock)
  236. def _on_connect_handler(
  237. self, client: MQTToTClient, _: Any, flags: dict[str, Any], rc: int
  238. ) -> None:
  239. if rc != 0:
  240. err = pmc.connack_string(rc)
  241. self.log.error("MQTT Connection Error: %s (%d)", err, rc)
  242. if rc == pmc.CONNACK_REFUSED_NOT_AUTHORIZED:
  243. self._disconnect_error = MQTTConnectionUnauthorized()
  244. self.disconnect()
  245. return
  246. self._loop.create_task(self._post_connect())
  247. def _on_disconnect_handler(self, client: MQTToTClient, _: Any, rc: int) -> None:
  248. err_str = "Generic error." if rc == pmc.MQTT_ERR_NOMEM else pmc.error_string(rc)
  249. self.log.debug(f"MQTT disconnection code %d: %s", rc, err_str)
  250. self._clear_response_waiters()
  251. async def _post_connect(self) -> None:
  252. await self._dispatch(Connect())
  253. self.log.debug("Re-subscribing to things after connect")
  254. if self._graphql_subs:
  255. res = await self.graphql_subscribe(self._graphql_subs)
  256. self.log.trace("GraphQL subscribe response: %s", res)
  257. if self._skywalker_subs:
  258. res = await self.skywalker_subscribe(self._skywalker_subs)
  259. self.log.trace("Skywalker subscribe response: %s", res)
  260. if self._iris_seq_id:
  261. retry = 0
  262. while True:
  263. try:
  264. await self.iris_subscribe(self._iris_seq_id, self._iris_snapshot_at_ms)
  265. break
  266. except (asyncio.TimeoutError, IrisSubscribeError) as e:
  267. self.log.exception("Error requesting iris subscribe")
  268. retry += 1
  269. if retry >= 5 or isinstance(e, IrisSubscribeError):
  270. self._disconnect_error = e
  271. self.disconnect()
  272. break
  273. await asyncio.sleep(5)
  274. self.log.debug("Retrying iris subscribe")
  275. def _on_publish_handler(self, client: MQTToTClient, _: Any, mid: int) -> None:
  276. try:
  277. waiter = self._publish_waiters[mid]
  278. except KeyError:
  279. self.log.trace(f"Got publish confirmation for {mid}, but no waiters")
  280. return
  281. self.log.trace(f"Got publish confirmation for {mid}")
  282. waiter.set_result(None)
  283. # region Incoming event parsing
  284. def _parse_direct_thread_path(self, path: str) -> dict:
  285. try:
  286. blank, direct_v2, threads, thread_id, *rest = path.split("/")
  287. except (ValueError, IndexError) as e:
  288. self.log.debug(f"Got {e!r} while parsing path {path}")
  289. raise
  290. if (blank, direct_v2, threads) != ("", "direct_v2", "threads"):
  291. self.log.debug(f"Got unexpected first parts in direct thread path {path}")
  292. raise ValueError("unexpected first three parts in _parse_direct_thread_path")
  293. additional = {"thread_id": thread_id}
  294. if rest:
  295. subitem_key = rest[0]
  296. if subitem_key == "approval_required_for_new_members":
  297. additional["approval_required_for_new_members"] = True
  298. elif subitem_key == "participants" and len(rest) > 2 and rest[2] == "has_seen":
  299. additional["has_seen"] = int(rest[1])
  300. elif subitem_key == "items":
  301. additional["item_id"] = rest[1]
  302. if len(rest) > 4 and rest[2] == "reactions":
  303. additional["reaction_type"] = ReactionType(rest[3])
  304. additional["reaction_user_id"] = int(rest[4])
  305. elif subitem_key in "admin_user_ids":
  306. additional["admin_user_id"] = int(rest[1])
  307. elif subitem_key == "activity_indicator_id":
  308. additional["activity_indicator_id"] = rest[1]
  309. self.log.trace("Parsed path %s -> %s", path, additional)
  310. return additional
  311. def _on_messager_sync_item(self, part: IrisPayloadData, parsed_item: IrisPayload) -> bool:
  312. if part.path.startswith("/direct_v2/threads/"):
  313. raw_message = {
  314. "path": part.path,
  315. "op": part.op,
  316. **self._parse_direct_thread_path(part.path),
  317. }
  318. try:
  319. json_value = json.loads(part.value)
  320. if "reaction_type" in raw_message:
  321. self.log.trace("Treating %s as new reaction data", json_value)
  322. raw_message["new_reaction"] = json_value
  323. json_value["sender_id"] = raw_message.pop("reaction_user_id")
  324. json_value["type"] = raw_message.pop("reaction_type")
  325. json_value["client_context"] = parsed_item.mutation_token
  326. if part.op == Operation.REMOVE:
  327. json_value["emoji"] = None
  328. json_value["timestamp"] = None
  329. else:
  330. raw_message = {
  331. **raw_message,
  332. **json_value,
  333. }
  334. except (json.JSONDecodeError, TypeError):
  335. raw_message["value"] = part.value
  336. message = MessageSyncMessage.deserialize(raw_message)
  337. evt = MessageSyncEvent(iris=parsed_item, message=message)
  338. elif part.path.startswith("/direct_v2/inbox/threads/"):
  339. raw_message = {
  340. "path": part.path,
  341. "op": part.op,
  342. **json.loads(part.value),
  343. }
  344. evt = ThreadSyncEvent.deserialize(raw_message)
  345. else:
  346. self.log.warning(f"Unsupported path {part.path}")
  347. return False
  348. self._outgoing_events.put_nowait(evt)
  349. return True
  350. def _on_message_sync(self, payload: bytes) -> None:
  351. parsed = json.loads(payload.decode("utf-8"))
  352. self.log.trace("Got message sync event: %s", parsed)
  353. has_items = False
  354. for sync_item in parsed:
  355. parsed_item = IrisPayload.deserialize(sync_item)
  356. if self._iris_seq_id < parsed_item.seq_id:
  357. self.log.trace(f"Got new seq_id: {parsed_item.seq_id}")
  358. self._iris_seq_id = parsed_item.seq_id
  359. self._iris_snapshot_at_ms = int(time.time() * 1000)
  360. asyncio.create_task(
  361. self._dispatch(NewSequenceID(self._iris_seq_id, self._iris_snapshot_at_ms))
  362. )
  363. for part in parsed_item.data:
  364. has_items = self._on_messager_sync_item(part, parsed_item) or has_items
  365. if has_items and not self._event_dispatcher_task:
  366. self._event_dispatcher_task = asyncio.create_task(self._dispatcher_loop())
  367. def _on_pubsub(self, payload: bytes) -> None:
  368. parsed_thrift = IncomingMessage.from_thrift(payload)
  369. self.log.trace(f"Got pubsub event {parsed_thrift.topic} / {parsed_thrift.payload}")
  370. message = PubsubPayload.parse_json(parsed_thrift.payload)
  371. for data in message.data:
  372. match = ACTIVITY_INDICATOR_REGEX.match(data.path)
  373. if match:
  374. evt = PubsubEvent(
  375. data=data,
  376. base=message,
  377. thread_id=match.group(1),
  378. activity_indicator_id=match.group(2),
  379. )
  380. self._loop.create_task(self._dispatch(evt))
  381. elif not data.double_publish:
  382. self.log.debug("Pubsub: no activity indicator on data: %s", data)
  383. else:
  384. self.log.debug("Pubsub: double publish: %s", data.path)
  385. def _parse_realtime_sub_item(self, topic: str | GraphQLQueryID, raw: dict) -> Iterable[Any]:
  386. if topic == GraphQLQueryID.APP_PRESENCE:
  387. yield AppPresenceEventPayload.deserialize(raw).presence_event
  388. elif topic == GraphQLQueryID.ZERO_PROVISION:
  389. yield RealtimeZeroProvisionPayload.deserialize(raw).zero_product_provisioning_event
  390. elif topic == GraphQLQueryID.CLIENT_CONFIG_UPDATE:
  391. yield ClientConfigUpdatePayload.deserialize(raw).client_config_update_event
  392. elif topic == GraphQLQueryID.LIVE_REALTIME_COMMENTS:
  393. yield LiveVideoCommentPayload.deserialize(raw).live_video_comment_event
  394. elif topic == "direct":
  395. event = raw["event"]
  396. for item in raw["data"]:
  397. yield RealtimeDirectEvent.deserialize(
  398. {
  399. "event": event,
  400. **self._parse_direct_thread_path(item["path"]),
  401. **item,
  402. }
  403. )
  404. def _on_realtime_sub(self, payload: bytes) -> None:
  405. parsed_thrift = IncomingMessage.from_thrift(payload)
  406. try:
  407. topic = GraphQLQueryID(parsed_thrift.topic)
  408. except ValueError:
  409. topic = parsed_thrift.topic
  410. self.log.trace(f"Got realtime sub event {topic} / {parsed_thrift.payload}")
  411. allowed = (
  412. "direct",
  413. GraphQLQueryID.APP_PRESENCE,
  414. GraphQLQueryID.ZERO_PROVISION,
  415. GraphQLQueryID.CLIENT_CONFIG_UPDATE,
  416. GraphQLQueryID.LIVE_REALTIME_COMMENTS,
  417. )
  418. if topic not in allowed:
  419. return
  420. parsed_json = json.loads(parsed_thrift.payload)
  421. for evt in self._parse_realtime_sub_item(topic, parsed_json):
  422. self._loop.create_task(self._dispatch(evt))
  423. def _handle_send_response(self, message: pmc.MQTTMessage) -> None:
  424. data = json.loads(message.payload.decode("utf-8"))
  425. try:
  426. ccid = data["payload"]["client_context"]
  427. except KeyError:
  428. self.log.warning(
  429. "Didn't find client_context in send message response: %s", message.payload
  430. )
  431. ccid = self._message_response_waiter_id
  432. else:
  433. if ccid != self._message_response_waiter_id:
  434. self.log.error(
  435. "Mismatching client_context in send message response (%s != %s)",
  436. ccid,
  437. self._message_response_waiter_id,
  438. )
  439. return
  440. if self._message_response_waiter and not self._message_response_waiter.done():
  441. self.log.debug("Got response to %s: %s", ccid, message.payload)
  442. self._message_response_waiter.set_result(message)
  443. self._message_response_waiter = None
  444. else:
  445. self.log.warning("Didn't find task waiting for response %s", message.payload)
  446. def _on_message_handler(self, client: MQTToTClient, _: Any, message: pmc.MQTTMessage) -> None:
  447. try:
  448. topic = RealtimeTopic.decode(message.topic)
  449. # Instagram Android MQTT messages are always compressed
  450. message.payload = zlib.decompress(message.payload)
  451. if topic == RealtimeTopic.MESSAGE_SYNC:
  452. self._on_message_sync(message.payload)
  453. elif topic == RealtimeTopic.PUBSUB:
  454. self._on_pubsub(message.payload)
  455. elif topic == RealtimeTopic.REALTIME_SUB:
  456. self._on_realtime_sub(message.payload)
  457. elif topic == RealtimeTopic.SEND_MESSAGE_RESPONSE:
  458. self._handle_send_response(message)
  459. else:
  460. try:
  461. waiter = self._response_waiters.pop(topic)
  462. except KeyError:
  463. self.log.debug(
  464. "No handler for MQTT message in %s: %s", topic.value, message.payload
  465. )
  466. else:
  467. self.log.trace("Got response %s: %s", topic.value, message.payload)
  468. waiter.set_result(message)
  469. except Exception:
  470. self.log.exception("Error in incoming MQTT message handler")
  471. self.log.trace("Errored MQTT payload: %s", message.payload)
  472. # endregion
  473. async def _reconnect(self) -> None:
  474. try:
  475. self.log.trace("Trying to reconnect to MQTT")
  476. self._client.reconnect()
  477. except (SocketError, OSError, pmc.WebsocketConnectionError) as e:
  478. # TODO custom class
  479. raise MQTTNotLoggedIn("MQTT reconnection failed") from e
  480. def add_event_handler(
  481. self, evt_type: Type[T], handler: Callable[[T], Awaitable[None]]
  482. ) -> None:
  483. self._event_handlers[evt_type].append(handler)
  484. async def _dispatch(self, evt: T) -> None:
  485. for handler in self._event_handlers[type(evt)]:
  486. try:
  487. await handler(evt)
  488. except Exception:
  489. self.log.exception(f"Error in {type(evt).__name__} handler")
  490. def disconnect(self) -> None:
  491. self._client.disconnect()
  492. async def _dispatcher_loop(self) -> None:
  493. loop_id = f"{hex(id(self))}#{time.monotonic()}"
  494. self.log.debug(f"Dispatcher loop {loop_id} starting")
  495. try:
  496. while True:
  497. evt = await self._outgoing_events.get()
  498. await asyncio.shield(self._dispatch(evt))
  499. except asyncio.CancelledError:
  500. tasks = self._outgoing_events
  501. self._outgoing_events = asyncio.Queue()
  502. if not tasks.empty():
  503. self.log.debug(
  504. f"Dispatcher loop {loop_id} stopping after dispatching {tasks.qsize()} events"
  505. )
  506. while not tasks.empty():
  507. await self._dispatch(tasks.get_nowait())
  508. raise
  509. finally:
  510. self.log.debug(f"Dispatcher loop {loop_id} stopped")
  511. async def listen(
  512. self,
  513. graphql_subs: set[str] | None = None,
  514. skywalker_subs: set[str] | None = None,
  515. seq_id: int = None,
  516. snapshot_at_ms: int = None,
  517. retry_limit: int = 5,
  518. ) -> None:
  519. self._graphql_subs = graphql_subs or set()
  520. self._skywalker_subs = skywalker_subs or set()
  521. self._iris_seq_id = seq_id
  522. self._iris_snapshot_at_ms = snapshot_at_ms
  523. self.log.debug("Connecting to Instagram MQTT")
  524. await self._reconnect()
  525. connection_retries = 0
  526. while True:
  527. try:
  528. await asyncio.sleep(1)
  529. except asyncio.CancelledError:
  530. self.disconnect()
  531. # this might not be necessary
  532. self._client.loop_misc()
  533. break
  534. rc = self._client.loop_misc()
  535. # If disconnect() has been called
  536. # Beware, internal API, may have to change this to something more stable!
  537. if self._client._state == pmc.mqtt_cs_disconnecting:
  538. break # Stop listening
  539. if rc != pmc.MQTT_ERR_SUCCESS:
  540. # If known/expected error
  541. if rc == pmc.MQTT_ERR_CONN_LOST:
  542. await self._dispatch(Disconnect(reason="Connection lost, retrying"))
  543. elif rc == pmc.MQTT_ERR_NOMEM:
  544. # This error is wrongly classified
  545. # See https://github.com/eclipse/paho.mqtt.python/issues/340
  546. await self._dispatch(Disconnect(reason="Connection lost, retrying"))
  547. elif rc == pmc.MQTT_ERR_CONN_REFUSED:
  548. raise MQTTNotLoggedIn("MQTT connection refused")
  549. elif rc == pmc.MQTT_ERR_NO_CONN:
  550. if connection_retries > retry_limit:
  551. raise MQTTNotConnected(f"Connection failed {connection_retries} times")
  552. if self.proxy_handler.update_proxy_url():
  553. self.setup_proxy()
  554. await self._dispatch(ProxyUpdate())
  555. sleep = connection_retries * 2
  556. await self._dispatch(
  557. Disconnect(
  558. reason="MQTT Error: no connection, retrying "
  559. f"in {connection_retries} seconds"
  560. )
  561. )
  562. await asyncio.sleep(sleep)
  563. else:
  564. err = pmc.error_string(rc)
  565. self.log.error("MQTT Error: %s", err)
  566. await self._dispatch(Disconnect(reason=f"MQTT Error: {err}, retrying"))
  567. await self._reconnect()
  568. connection_retries += 1
  569. else:
  570. connection_retries = 0
  571. if self._event_dispatcher_task:
  572. self._event_dispatcher_task.cancel()
  573. self._event_dispatcher_task = None
  574. if self._disconnect_error:
  575. self.log.info("disconnect_error is set, raising and clearing variable")
  576. err = self._disconnect_error
  577. self._disconnect_error = None
  578. raise err
  579. # region Basic outgoing MQTT
  580. def publish(self, topic: RealtimeTopic, payload: str | bytes | dict) -> asyncio.Future:
  581. if isinstance(payload, dict):
  582. payload = json.dumps(payload)
  583. if isinstance(payload, str):
  584. payload = payload.encode("utf-8")
  585. self.log.trace(f"Publishing message in {topic.value} ({topic.encoded}): {payload}")
  586. payload = zlib.compress(payload, level=9)
  587. info = self._client.publish(topic.encoded, payload, qos=1)
  588. self.log.trace(f"Published message ID: {info.mid}")
  589. fut = asyncio.Future()
  590. self._publish_waiters[info.mid] = fut
  591. return fut
  592. async def request(
  593. self,
  594. topic: RealtimeTopic,
  595. response: RealtimeTopic,
  596. payload: str | bytes | dict,
  597. timeout: int | None = None,
  598. ) -> pmc.MQTTMessage:
  599. async with self._response_waiter_locks[response]:
  600. fut = asyncio.Future()
  601. self._response_waiters[response] = fut
  602. await self.publish(topic, payload)
  603. self.log.trace(
  604. f"Request published to {topic.value}, waiting for response {response.name}"
  605. )
  606. return await asyncio.wait_for(fut, timeout)
  607. async def iris_subscribe(self, seq_id: int, snapshot_at_ms: int) -> None:
  608. self.log.debug(f"Requesting iris subscribe {seq_id}/{snapshot_at_ms}")
  609. resp = await self.request(
  610. RealtimeTopic.SUB_IRIS,
  611. RealtimeTopic.SUB_IRIS_RESPONSE,
  612. {
  613. "seq_id": seq_id,
  614. "snapshot_at_ms": snapshot_at_ms,
  615. "snapshot_app_version": self.state.application.APP_VERSION,
  616. "timezone_offset": int(self.state.device.timezone_offset),
  617. "subscription_type": "message",
  618. },
  619. timeout=20,
  620. )
  621. self.log.debug("Iris subscribe response: %s", resp.payload.decode("utf-8"))
  622. resp_dict = json.loads(resp.payload.decode("utf-8"))
  623. if resp_dict["error_type"] and resp_dict["error_message"]:
  624. raise IrisSubscribeError(resp_dict["error_type"], resp_dict["error_message"])
  625. latest_seq_id = resp_dict.get("latest_seq_id")
  626. if latest_seq_id > self._iris_seq_id:
  627. self.log.info(f"Latest sequence ID is {latest_seq_id}, catching up from {seq_id}")
  628. self._iris_seq_id = latest_seq_id
  629. self._iris_snapshot_at_ms = resp_dict.get("subscribed_at_ms", int(time.time() * 1000))
  630. asyncio.create_task(
  631. self._dispatch(NewSequenceID(self._iris_seq_id, self._iris_snapshot_at_ms))
  632. )
  633. def graphql_subscribe(self, subs: set[str]) -> asyncio.Future:
  634. self._graphql_subs |= subs
  635. return self.publish(RealtimeTopic.REALTIME_SUB, {"sub": list(subs)})
  636. def graphql_unsubscribe(self, subs: set[str]) -> asyncio.Future:
  637. self._graphql_subs -= subs
  638. return self.publish(RealtimeTopic.REALTIME_SUB, {"unsub": list(subs)})
  639. def skywalker_subscribe(self, subs: set[str]) -> asyncio.Future:
  640. self._skywalker_subs |= subs
  641. return self.publish(RealtimeTopic.PUBSUB, {"sub": list(subs)})
  642. def skywalker_unsubscribe(self, subs: set[str]) -> asyncio.Future:
  643. self._skywalker_subs -= subs
  644. return self.publish(RealtimeTopic.PUBSUB, {"unsub": list(subs)})
  645. # endregion
  646. # region Actually sending messages and stuff
  647. async def send_foreground_state(self, state: ForegroundStateConfig) -> None:
  648. self.log.debug("Updating foreground state: %s", state)
  649. await self.publish(
  650. RealtimeTopic.FOREGROUND_STATE, zlib.compress(state.to_thrift(), level=9)
  651. )
  652. if state.keep_alive_timeout:
  653. self._client._keepalive = state.keep_alive_timeout
  654. async def send_command(
  655. self,
  656. thread_id: str,
  657. action: ThreadAction,
  658. client_context: str | None = None,
  659. **kwargs: Any,
  660. ) -> CommandResponse | None:
  661. client_context = client_context or self.state.gen_client_context()
  662. req = {
  663. "thread_id": thread_id,
  664. "client_context": client_context,
  665. "offline_threading_id": client_context,
  666. "action": action.value,
  667. # "device_id": self.state.cookies["ig_did"],
  668. **kwargs,
  669. }
  670. lock_start = time.monotonic()
  671. async with self._message_response_waiter_lock:
  672. lock_wait_dur = time.monotonic() - lock_start
  673. if lock_wait_dur > 1:
  674. self.log.debug(f"Waited {lock_wait_dur:.3f} seconds to send {client_context}")
  675. fut = self._message_response_waiter = asyncio.Future()
  676. self._message_response_waiter_id = client_context
  677. await self.publish(RealtimeTopic.SEND_MESSAGE, req)
  678. self.log.trace(
  679. f"Request published to {RealtimeTopic.SEND_MESSAGE}, "
  680. f"waiting for response {RealtimeTopic.SEND_MESSAGE_RESPONSE}"
  681. )
  682. try:
  683. resp = await asyncio.wait_for(fut, timeout=30000)
  684. except asyncio.TimeoutError:
  685. self.log.error(f"Request with ID {client_context} timed out!")
  686. raise
  687. return CommandResponse.parse_json(resp.payload.decode("utf-8"))
  688. def send_item(
  689. self,
  690. thread_id: str,
  691. item_type: ThreadItemType,
  692. shh_mode: bool = False,
  693. client_context: str | None = None,
  694. **kwargs: Any,
  695. ) -> Awaitable[CommandResponse]:
  696. return self.send_command(
  697. thread_id,
  698. item_type=item_type.value,
  699. is_shh_mode=str(int(shh_mode)),
  700. action=ThreadAction.SEND_ITEM,
  701. client_context=client_context,
  702. **kwargs,
  703. )
  704. def send_hashtag(
  705. self,
  706. thread_id: str,
  707. hashtag: str,
  708. text: str = "",
  709. shh_mode: bool = False,
  710. client_context: str | None = None,
  711. ) -> Awaitable[CommandResponse]:
  712. return self.send_item(
  713. thread_id,
  714. text=text,
  715. item_id=hashtag,
  716. shh_mode=shh_mode,
  717. item_type=ThreadItemType.HASHTAG,
  718. client_context=client_context,
  719. )
  720. def send_like(
  721. self, thread_id: str, shh_mode: bool = False, client_context: str | None = None
  722. ) -> Awaitable[CommandResponse]:
  723. return self.send_item(
  724. thread_id,
  725. shh_mode=shh_mode,
  726. item_type=ThreadItemType.LIKE,
  727. client_context=client_context,
  728. )
  729. def send_location(
  730. self,
  731. thread_id: str,
  732. venue_id: str,
  733. text: str = "",
  734. shh_mode: bool = False,
  735. client_context: str | None = None,
  736. ) -> Awaitable[CommandResponse]:
  737. return self.send_item(
  738. thread_id,
  739. text=text,
  740. item_id=venue_id,
  741. shh_mode=shh_mode,
  742. item_type=ThreadItemType.LOCATION,
  743. client_context=client_context,
  744. )
  745. def send_media(
  746. self,
  747. thread_id: str,
  748. media_id: str,
  749. text: str = "",
  750. shh_mode: bool = False,
  751. client_context: str | None = None,
  752. ) -> Awaitable[CommandResponse]:
  753. return self.send_item(
  754. thread_id,
  755. text=text,
  756. media_id=media_id,
  757. shh_mode=shh_mode,
  758. item_type=ThreadItemType.MEDIA_SHARE,
  759. client_context=client_context,
  760. )
  761. def send_profile(
  762. self,
  763. thread_id: str,
  764. user_id: str,
  765. text: str = "",
  766. shh_mode: bool = False,
  767. client_context: str | None = None,
  768. ) -> Awaitable[CommandResponse]:
  769. return self.send_item(
  770. thread_id,
  771. text=text,
  772. item_id=user_id,
  773. shh_mode=shh_mode,
  774. item_type=ThreadItemType.PROFILE,
  775. client_context=client_context,
  776. )
  777. def send_reaction(
  778. self,
  779. thread_id: str,
  780. emoji: str,
  781. item_id: str,
  782. reaction_status: ReactionStatus = ReactionStatus.CREATED,
  783. target_item_type: ThreadItemType = ThreadItemType.TEXT,
  784. shh_mode: bool = False,
  785. client_context: str | None = None,
  786. ) -> Awaitable[CommandResponse]:
  787. return self.send_item(
  788. thread_id,
  789. reaction_status=reaction_status.value,
  790. node_type="item",
  791. reaction_type="like",
  792. target_item_type=target_item_type.value,
  793. emoji=emoji,
  794. item_id=item_id,
  795. reaction_action_source="double_tap",
  796. shh_mode=shh_mode,
  797. item_type=ThreadItemType.REACTION,
  798. client_context=client_context,
  799. )
  800. def send_user_story(
  801. self,
  802. thread_id: str,
  803. media_id: str,
  804. text: str = "",
  805. shh_mode: bool = False,
  806. client_context: str | None = None,
  807. ) -> Awaitable[CommandResponse]:
  808. return self.send_item(
  809. thread_id,
  810. text=text,
  811. item_id=media_id,
  812. shh_mode=shh_mode,
  813. item_type=ThreadItemType.REEL_SHARE,
  814. client_context=client_context,
  815. )
  816. def send_text(
  817. self,
  818. thread_id: str,
  819. text: str = "",
  820. urls: list[str] | None = None,
  821. shh_mode: bool = False,
  822. client_context: str | None = None,
  823. replied_to_item_id: str | None = None,
  824. replied_to_client_context: str | None = None,
  825. ) -> Awaitable[CommandResponse]:
  826. args = {
  827. "text": text,
  828. }
  829. item_type = ThreadItemType.TEXT
  830. if urls is not None:
  831. args = {
  832. "link_text": text,
  833. "link_urls": json.dumps(urls or []),
  834. }
  835. item_type = ThreadItemType.LINK
  836. return self.send_item(
  837. thread_id,
  838. **args,
  839. shh_mode=shh_mode,
  840. item_type=item_type,
  841. client_context=client_context,
  842. replied_to_item_id=replied_to_item_id,
  843. replied_to_client_context=replied_to_client_context,
  844. )
  845. def mark_seen(
  846. self, thread_id: str, item_id: str, client_context: str | None = None
  847. ) -> Awaitable[None]:
  848. return self.send_command(
  849. thread_id,
  850. item_id=item_id,
  851. action=ThreadAction.MARK_SEEN,
  852. client_context=client_context,
  853. )
  854. def mark_visual_item_seen(
  855. self, thread_id: str, item_id: str, client_context: str | None = None
  856. ) -> Awaitable[CommandResponse]:
  857. return self.send_command(
  858. thread_id,
  859. item_id=item_id,
  860. action=ThreadAction.MARK_VISUAL_ITEM_SEEN,
  861. client_context=client_context,
  862. )
  863. def indicate_activity(
  864. self,
  865. thread_id: str,
  866. activity_status: TypingStatus = TypingStatus.TEXT,
  867. client_context: str | None = None,
  868. ) -> Awaitable[CommandResponse]:
  869. return self.send_command(
  870. thread_id,
  871. activity_status=activity_status.value,
  872. action=ThreadAction.INDICATE_ACTIVITY,
  873. client_context=client_context,
  874. )
  875. # endregion