matrix-appservice-kakaotalk/matrix_appservice_kakaotalk/commands/auth.py

201 lines
7.0 KiB
Python

# matrix-appservice-kakaotalk - A Matrix-KakaoTalk puppeting bridge.
# Copyright (C) 2022 Tulir Asokan, Andrew Ferrazzutti
#
# 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/>.
import time
from yarl import URL
from mautrix.bridge.commands import HelpSection, command_handler
from mautrix.errors import MForbidden
from mautrix.util.signed_token import sign_token
from ..kt.client import Client as KakaoTalkClient
from ..kt.client.errors import DeviceVerificationRequired, IncorrectPasscode, IncorrectPassword, CommandException
from .. import puppet as pu
from .typehint import CommandEvent
SECTION_AUTH = HelpSection("Authentication", 10, "")
web_unsupported = (
"This instance of the KakaoTalk bridge does not support the web-based login interface"
)
alternative_web_login = (
"Alternatively, you may use [the web-based login interface]({url}) "
"to prevent the bridge and homeserver from seeing your password"
)
forced_web_login = (
"This instance of the KakaoTalk bridge does not allow in-Matrix login. "
"Please use [the web-based login interface]({url})."
)
send_password = "Please send your password here to log in"
missing_email = "Please use `$cmdprefix+sp login <email>` to log in here"
try_again_or_cancel = "Try again, or say `$cmdprefix+sp cancel` to give up."
@command_handler(
needs_auth=False,
management_only=True,
help_section=SECTION_AUTH,
help_text="Log in to KakaoTalk",
help_args="[_email_]",
)
async def login(evt: CommandEvent) -> None:
if await evt.sender.is_logged_in():
await evt.reply("You're already logged in")
return
email = evt.args[0] if len(evt.args) > 0 else None
if email:
evt.sender.command_status = {
"action": "Login",
"room_id": evt.room_id,
"next": enter_password,
"email": evt.args[0],
}
if evt.bridge.public_website:
external_url = URL(evt.config["appservice.public.external"])
token = sign_token(
evt.bridge.public_website.secret_key,
{
"mxid": evt.sender.mxid,
"expiry": int(time.time()) + 30 * 60,
},
)
url = (external_url / "login.html").with_fragment(token)
if not evt.config["appservice.public.allow_matrix_login"]:
await evt.reply(forced_web_login.format(url=url))
elif email:
await evt.reply(f"{send_password}. {alternative_web_login.format(url=url)}.")
else:
await evt.reply(f"{missing_email}. {alternative_web_login.format(url=url)}.")
elif not email:
await evt.reply(f"{missing_email}. {web_unsupported}.")
else:
await evt.reply(f"{send_password}. {web_unsupported}.")
async def enter_password(evt: CommandEvent) -> None:
try:
await evt.az.intent.redact(evt.room_id, evt.event_id)
except MForbidden:
pass
assert evt.sender.command_status
req = {
"uuid": await evt.sender.get_uuid(),
"form": {
"email": evt.sender.command_status["email"],
"password": evt.content.body,
}
}
try:
await _do_login(evt, req)
except DeviceVerificationRequired:
await evt.reply(
"Open KakaoTalk on your smartphone. It should show a device registration passcode. "
"Enter that passcode here."
)
evt.sender.command_status = {
"action": "Login",
"room_id": evt.room_id,
"next": enter_dv_code,
"req": req,
}
except IncorrectPassword:
await evt.reply(f"Incorrect password. {try_again_or_cancel}")
#except OAuthException as e:
# await evt.reply(f"Error from KakaoTalk:\n\n> {e}")
except Exception as e:
await _handle_login_failure(evt, e)
async def enter_dv_code(evt: CommandEvent) -> None:
assert evt.sender.command_status
req: dict = evt.sender.command_status["req"]
passcode = evt.content.body
await evt.mark_read()
try:
await KakaoTalkClient.register_device(passcode, **req)
await _do_login(evt, req)
except IncorrectPasscode:
await evt.reply(f"Incorrect device registration passcode. {try_again_or_cancel}")
#except OAuthException as e:
# await evt.reply(f"Error from KakaoTalk:\n\n> {e}")
except Exception as e:
await _handle_login_failure(evt, e)
async def _do_login(evt: CommandEvent, req: dict) -> None:
oauth_credential = await KakaoTalkClient.login(**req)
await evt.sender.on_logged_in(oauth_credential)
evt.sender.command_status = None
await evt.reply("Successfully logged in")
async def _handle_login_failure(evt: CommandEvent, e: Exception) -> None:
evt.sender.command_status = None
if isinstance(e, CommandException):
message = "Failed to log in"
evt.log.error(message)
else:
message = "Error while logging in"
evt.log.exception(message)
await evt.reply(f"{message}: {e}")
@command_handler(
needs_auth=True,
help_section=SECTION_AUTH,
help_text="Log out of KakaoTalk (and optionally change your virtual device ID for next login)",
help_args="[--reset-device]",
)
async def logout(evt: CommandEvent) -> None:
if len(evt.args) >= 1:
if evt.args[0] == "--reset-device":
reset_device = True
else:
await evt.reply("**Usage:** `$cmdprefix+sp logout [--reset-device]`")
return
else:
reset_device = False
puppet = await pu.Puppet.get_by_ktid(evt.sender.ktid)
await evt.sender.logout(reset_device=reset_device)
if puppet.is_real_user:
await puppet.switch_mxid(None, None)
message = "Successfully logged out"
if reset_device:
message += (
", and your next login will use a different device ID.\n\n"
"The old device must be manually de-registered from the KakaoTalk app."
)
await evt.reply(message)
@command_handler(needs_auth=False, help_section=SECTION_AUTH, help_text="Change your virtual device ID for next login")
async def reset_device(evt: CommandEvent) -> None:
if await evt.sender.is_logged_in():
await evt.reply("This command requires you to be logged out.")
else:
await evt.mark_read()
await evt.sender.logout(reset_device=True)
await evt.reply(
"Your next login will use a different device ID.\n\n"
"The old device must be manually de-registered from the KakaoTalk app."
)