76 lines
No EOL
2.8 KiB
Python
76 lines
No EOL
2.8 KiB
Python
import asyncio
|
|
|
|
from aiograpi import Client
|
|
from aiograpi.exceptions import (
|
|
BadPassword,
|
|
ChallengeRequired,
|
|
FeedbackRequired,
|
|
TwoFactorRequired,
|
|
)
|
|
from slidge import BaseGateway
|
|
from slidge.command.register import RegistrationType, TwoFactorNotRequired
|
|
from slidge.db import GatewayUser
|
|
from slixmpp import JID
|
|
|
|
from .utils import get_session_file
|
|
|
|
|
|
class Gateway(BaseGateway):
|
|
COMPONENT_NAME = "Instagram (Slidge)"
|
|
COMPONENT_TYPE = "instagram"
|
|
COMPONENT_AVATAR = ("https://upload.wikimedia.org/wikipedia/commons/thumb/e/e7/"
|
|
"Instagram_logo_2016.svg/640px-Instagram_logo_2016.svg.png")
|
|
|
|
ROSTER_GROUP = "instagram"
|
|
|
|
REGISTRATION_INSTRUCTIONS = "Enter instagram credentials"
|
|
REGISTRATION_TYPE = RegistrationType.TWO_FACTOR_CODE
|
|
|
|
GROUPS = False
|
|
|
|
def __init__(self):
|
|
super().__init__()
|
|
self.instagram_client = dict[str, Client]()
|
|
|
|
async def validate(self, user_jid: JID, registration_form: dict[str, str | None]):
|
|
session_file = get_session_file(user_jid.bare)
|
|
client = Client()
|
|
try:
|
|
await client.login(
|
|
registration_form["username"],
|
|
registration_form["password"]
|
|
)
|
|
except TwoFactorRequired:
|
|
self.instagram_client[user_jid.bare] = client
|
|
except BadPassword as e:
|
|
raise ValueError("Bad Password", e, e.args)
|
|
except ChallengeRequired as e:
|
|
raise ValueError("Browser Challenge Required", e, e.args)
|
|
except FeedbackRequired as e:
|
|
raise ValueError("Action moderated, account may be blocked", e, e.args)
|
|
except Exception as e:
|
|
raise ValueError("Could not authenticate: %s - %s", e, e.args)
|
|
else:
|
|
await self._save_settings(client, session_file)
|
|
raise TwoFactorNotRequired
|
|
|
|
async def validate_two_factor_code(self, user: GatewayUser, code: str):
|
|
session_file = get_session_file(user.jid.bare)
|
|
client = self.instagram_client[user.jid.bare]
|
|
try:
|
|
if await client.login(
|
|
user.legacy_module_data["username"],
|
|
user.legacy_module_data["password"],
|
|
verification_code=code):
|
|
await self._save_settings(client, session_file)
|
|
except ChallengeRequired as e:
|
|
raise ValueError("Browser Challenge Required", e, e.args)
|
|
except FeedbackRequired as e:
|
|
raise ValueError("Action moderated, account may be blocked", e, e.args)
|
|
except Exception as e:
|
|
raise ValueError("Could not authenticate: %s - %s", e, e.args)
|
|
|
|
@staticmethod
|
|
async def _save_settings(client: Client, session_file: str):
|
|
if not await asyncio.to_thread(client.dump_settings, session_file):
|
|
raise IOError("Could not save session file: %s - %s", session_file) |