forked from fair/matrix-puppeteer-line
127 lines
5.4 KiB
Python
127 lines
5.4 KiB
Python
# matrix-puppeteer-line - A very hacky Matrix-LINE bridge based on running LINE's Chrome extension in Puppeteer
|
|
# Copyright (C) 2020-2021 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/>.
|
|
from typing import Optional, AsyncGenerator, Tuple
|
|
import io
|
|
|
|
import qrcode
|
|
import PIL as _
|
|
|
|
from mautrix.types import TextMessageEventContent, MediaMessageEventContent, MessageType, ImageInfo, EventID
|
|
from mautrix.bridge.commands import HelpSection, command_handler
|
|
|
|
from .typehint import CommandEvent
|
|
|
|
SECTION_AUTH = HelpSection("Authentication", 10, "")
|
|
|
|
|
|
async def login_prep(evt: CommandEvent, login_type: str) -> bool:
|
|
status = await evt.sender.client.start()
|
|
if status.is_logged_in:
|
|
await evt.reply("You're already logged in")
|
|
return False
|
|
|
|
if evt.sender.command_status is not None:
|
|
action = evt.sender.command_status["action"]
|
|
if action == "Login":
|
|
await evt.reply(
|
|
"A login is already in progress. Please follow the login instructions, "
|
|
"or use the `$cmdprefix+sp cancel` command to start over.")
|
|
else:
|
|
await evt.reply(f"Cannot login while a {action} command is active.")
|
|
return False
|
|
|
|
evt.sender.command_status = {
|
|
"action": "Login",
|
|
"login_type": login_type,
|
|
}
|
|
return True
|
|
|
|
async def login_do(evt: CommandEvent, gen: AsyncGenerator[Tuple[str, str], None]) -> None:
|
|
qr_event_id: Optional[EventID] = None
|
|
pin_event_id: Optional[EventID] = None
|
|
failure = False
|
|
async for item in gen:
|
|
if item[0] == "qr":
|
|
message = "Open LINE on your smartphone and scan this QR code:"
|
|
content = TextMessageEventContent(body=message, msgtype=MessageType.NOTICE)
|
|
content.set_reply(evt.event_id)
|
|
await evt.az.intent.send_message(evt.room_id, content)
|
|
|
|
url = item[1]
|
|
buffer = io.BytesIO()
|
|
image = qrcode.make(url)
|
|
size = image.pixel_size
|
|
image.save(buffer, "PNG")
|
|
qr = buffer.getvalue()
|
|
mxc = await evt.az.intent.upload_media(qr, "image/png", "login-qr.png", len(qr))
|
|
content = MediaMessageEventContent(body=url, url=mxc, msgtype=MessageType.IMAGE,
|
|
info=ImageInfo(mimetype="image/png", size=len(qr),
|
|
width=size, height=size))
|
|
if qr_event_id:
|
|
content.set_edit(qr_event_id)
|
|
await evt.az.intent.send_message(evt.room_id, content)
|
|
else:
|
|
qr_event_id = await evt.az.intent.send_message(evt.room_id, content)
|
|
elif item[0] == "pin":
|
|
pin = item[1]
|
|
message = f"Enter this PIN in LINE on your smartphone:\n{pin}"
|
|
content = TextMessageEventContent(body=message, msgtype=MessageType.NOTICE)
|
|
if pin_event_id:
|
|
content.set_edit(pin_event_id)
|
|
await evt.az.intent.send_message(evt.room_id, content)
|
|
else:
|
|
pin_event_id = await evt.az.intent.send_message(evt.room_id, content)
|
|
elif item[0] == "login_success":
|
|
await evt.reply("Successfully logged in, waiting for LINE to load...")
|
|
elif item[0] in ("login_failure", "error"):
|
|
# TODO Handle errors differently?
|
|
failure = True
|
|
reason = item[1]
|
|
if reason:
|
|
content = TextMessageEventContent(body=reason, msgtype=MessageType.NOTICE)
|
|
await evt.az.intent.send_message(evt.room_id, content)
|
|
# else: pass
|
|
|
|
if not failure and evt.sender.command_status:
|
|
await evt.reply("LINE loading complete")
|
|
await evt.sender.sync()
|
|
# else command was cancelled or failed. Don't post message about it, "cancel" command or failure did already
|
|
evt.sender.command_status = None
|
|
|
|
@command_handler(needs_auth=False, management_only=True, help_section=SECTION_AUTH,
|
|
help_text="Log into LINE via QR code")
|
|
async def login_qr(evt: CommandEvent) -> None:
|
|
if not await login_prep(evt, "qr"):
|
|
return
|
|
gen = evt.sender.client.login(evt.sender)
|
|
await login_do(evt, gen)
|
|
|
|
@command_handler(needs_auth=False, management_only=True, help_section=SECTION_AUTH,
|
|
help_text="Log into LINE via email/password",
|
|
help_args="<_email_> <_password_>")
|
|
async def login_email(evt: CommandEvent) -> None:
|
|
await evt.az.intent.redact(evt.room_id, evt.event_id)
|
|
if len(evt.args) != 2:
|
|
await evt.reply("Usage: `$cmdprefix+sp login-email <email> <password>`")
|
|
return
|
|
if not await login_prep(evt, "email"):
|
|
return
|
|
await evt.reply("Logging in...")
|
|
gen = evt.sender.client.login(
|
|
evt.sender,
|
|
login_data=dict(email=evt.args[0], password=evt.args[1]))
|
|
await login_do(evt, gen)
|