From 81c98e18c01f7ba2ba80f62e893f0e488fbd2801 Mon Sep 17 00:00:00 2001 From: DomNomNom Date: Sat, 15 Feb 2025 22:42:38 +1300 Subject: [PATCH] dump --- getbot/access_token_login.py | 96 +----------------------------- getbot/login.py | 110 +++++++++++++++++++++++++++++++++++ getbot/main.py | 58 +++++++++--------- getbot/verify_with_emoji.py | 29 +++++---- 4 files changed, 160 insertions(+), 133 deletions(-) create mode 100755 getbot/login.py diff --git a/getbot/access_token_login.py b/getbot/access_token_login.py index 56b4c39..d267ce3 100755 --- a/getbot/access_token_login.py +++ b/getbot/access_token_login.py @@ -1,71 +1,38 @@ #!/usr/bin/env python3 - import asyncio - import getpass - import json - import os - import sys - - import aiofiles - - from nio import AsyncClient, LoginResponse - - CONFIG_FILE = "credentials.json" - # Check out main() below to see how it's done. - - def write_details_to_disk(resp: LoginResponse, homeserver) -> None: - """Writes the required login details to disk so we can log in later without - using a password. - - Arguments: - resp {LoginResponse} -- the successful client login response. - homeserver -- URL of homeserver, e.g. "https://matrix.example.org" - """ - # open the config file in write-mode - with open(CONFIG_FILE, "w") as f: - # write the login details to disk - json.dump( { "homeserver": homeserver, # e.g. "https://matrix.example.org" - "user_id": resp.user_id, # e.g. "@user:example.org" - "device_id": resp.device_id, # device ID, 10 uppercase letters - "access_token": resp.access_token, # cryptogr. access token - }, - f, - ) - - async def main() -> None: # If there are no previously-saved credentials, we'll use the password @@ -75,114 +42,53 @@ async def main() -> None: print( "First time use. Did not find credential file. Asking for " - "homeserver, user, and password to create credential file." - ) - # homeserver = "https://matrix.example.org" - # homeserver = input(f"Enter your homeserver URL: [{homeserver}] ") - - # if not (homeserver.startswith("https://") or homeserver.startswith("http://")): - # homeserver = "https://" + homeserver - - # user_id = "@user:example.org" - # user_id = input(f"Enter your full user ID: [{user_id}] ") - - # device_name = "matrix-nio" - # device_name = input(f"Choose a name for this device: [{device_name}] ") - - # client = AsyncClient(homeserver, user_id) - # pw = getpass.getpass() - - homeserver = "https://matrix.domnomnom.com" user_id = "@getbot:matrix.domnomnom.com" room_id = "!sZpfYzLsRbnIOKJlPH:matrix.domnomnom.com" client = AsyncClient(homeserver, user_id) - # client.add_event_callback(message_callback, RoomMessageText) - - resp = await client.login(password) - - # check that we logged in successfully - if isinstance(resp, LoginResponse): - write_details_to_disk(resp, homeserver) - else: - print(f'homeserver = "{homeserver}"; user = "{user_id}"') - print(f"Failed to log in: {resp}") - sys.exit(1) - - print( - "Logged in using a password. Credentials were stored.", - "Try running the script again to login with credentials.", - ) - - # Otherwise the config file exists, so we'll use the stored credentials - else: - # open the file in read-only mode - async with aiofiles.open(CONFIG_FILE) as f: - contents = await f.read() - config = json.loads(contents) - client = AsyncClient(config["homeserver"]) - - client.access_token = config["access_token"] - client.user_id = config["user_id"] - client.device_id = config["device_id"] - - + room_id = config["room_id"] # Now we can send messages as the user - - - await client.room_send( - room_id, - message_type="m.room.message", - content={"msgtype": "m.text", "body": "Hello world!"}, - ) - print("Logged in using stored credentials. Sent a test message.") - - # Either way we're logged in here, too - await client.close() - - - asyncio.run(main()) diff --git a/getbot/login.py b/getbot/login.py new file mode 100755 index 0000000..aca3682 --- /dev/null +++ b/getbot/login.py @@ -0,0 +1,110 @@ +import asyncio +import getpass +import json +import os +import sys +import traceback + +import aiofiles + +from nio import ( + AsyncClient, + AsyncClientConfig, + KeyVerificationCancel, + KeyVerificationEvent, + KeyVerificationKey, + KeyVerificationMac, + KeyVerificationStart, + LocalProtocolError, + LoginResponse, + ToDeviceError, + ToDeviceMessage, + UnknownToDeviceEvent, +) + +CONFIG_FILE = "credentials.json" +STORE_PATH = "./store/" # local directory + +async def login() -> AsyncClient: + """Handle login with or without stored credentials.""" + # Configuration options for the AsyncClient + client_config = AsyncClientConfig( + max_limit_exceeded=0, + max_timeouts=0, + store_sync_tokens=True, + encryption_enabled=True, + ) + + # If there are no previously-saved credentials, we'll use the password + if not os.path.exists(CONFIG_FILE): + # print( + # "First time use. Did not find credential file. Asking for " + # "homeserver, user, and password to create credential file." + # ) + # homeserver = "https://matrix.example.org" + # homeserver = input(f"Enter your homeserver URL: [{homeserver}] ") + + # if not (homeserver.startswith("https://") or homeserver.startswith("http://")): + # homeserver = "https://" + homeserver + + # user_id = "@user:example.org" + # user_id = input(f"Enter your full user ID: [{user_id}] ") + + # device_name = "matrix-nio" + # device_name = input(f"Choose a name for this device: [{device_name}] ") + homeserver = "https://matrix.domnomnom.com" + user_id = "@getbot:matrix.domnomnom.com" + room_id = "!sZpfYzLsRbnIOKJlPH:matrix.domnomnom.com" + device_name = "whitebox-nio" + + + if not os.path.exists(STORE_PATH): + os.makedirs(STORE_PATH) + + # Initialize the matrix client + client = AsyncClient( + homeserver, + user_id, + store_path=STORE_PATH, + config=client_config, + ) + pw = getpass.getpass() + + resp = await client.login(password=pw, device_name=device_name) + + # check that we logged in successfully + if isinstance(resp, LoginResponse): + write_details_to_disk(resp, homeserver) + else: + print(f'homeserver = "{homeserver}"; user = "{user_id}"') + print(f"Failed to log in: {resp}") + sys.exit(1) + + print( + "Logged in using a password. Credentials were stored. " + "On next execution the stored login credentials will be used." + ) + + # Otherwise the config file exists, so we'll use the stored credentials + else: + # open the file in read-only mode + async with aiofiles.open(CONFIG_FILE, "r") as f: + contents = await f.read() + config = json.loads(contents) + # Initialize the matrix client based on credentials from file + client = AsyncClient( + config["homeserver"], + config["user_id"], + device_id=config["device_id"], + store_path=STORE_PATH, + config=client_config, + ) + + client.restore_login( + user_id=config["user_id"], + device_id=config["device_id"], + access_token=config["access_token"], + ) + print(f"Logged in using stored credentials as {client.user_id}. device_id={client.device_id}") + + return client diff --git a/getbot/main.py b/getbot/main.py index 7b90aec..a9f5472 100755 --- a/getbot/main.py +++ b/getbot/main.py @@ -1,50 +1,56 @@ import asyncio +import aiofiles.os from nio import AsyncClient, MatrixRoom, RoomMessageText +import asyncio +import getpass +import json +import os +import sys + +import aiofiles.os +import magic +from PIL import Image + +from nio import AsyncClient, LoginResponse, UploadResponse + +from login import login async def message_callback(room: MatrixRoom, event: RoomMessageText) -> None: - print( - f"Message received in room {room.display_name}\n" - - f"{room.user_name(event.sender)} | {event.body}" - + f"{room.user_name(event.sender) or event.sender} | {event.body}" ) async def main() -> None: - client = AsyncClient("https://matrix.domnomnom.com", "@getbot:matrix.domnomnom.com") + # async with aiofiles.open(CONFIG_FILE) as f: + # contents = await f.read() + # config = json.loads(contents) + # client = AsyncClient(config["homeserver"]) + # client.access_token = config["access_token"] + # client.user_id = config["user_id"] + # client.device_id = config["device_id"] + + # room_id = "!sZpfYzLsRbnIOKJlPH:matrix.domnomnom.com" + # image = "/home/get/getbot_profile_pic_256.png" + # # await send_image(client, room_id, image) + # if client.should_upload_keys: + # await client.keys_upload() + client = await login() + client.add_event_callback(message_callback, RoomMessageText) - print(await client.login(password)) - - # "Logged in as @alice:example.org device id: RANDOMDID" - - - # If you made a new room and haven't joined as that user, you can use - - # await client.join("your-room-id") - - - await client.room_send( - - room_id="!sZpfYzLsRbnIOKJlPH:matrix.domnomnom.com", - - message_type="m.room.message", - - content={"msgtype": "m.text", "body": "Hello world!"}, - - ) await client.sync_forever(timeout=30000) # milliseconds -asyncio.run(main()) +if __name__ == '__main__': + asyncio.run(main()) diff --git a/getbot/verify_with_emoji.py b/getbot/verify_with_emoji.py index aadd6fc..ef55cae 100755 --- a/getbot/verify_with_emoji.py +++ b/getbot/verify_with_emoji.py @@ -359,21 +359,26 @@ async def login() -> AsyncClient: # If there are no previously-saved credentials, we'll use the password if not os.path.exists(CONFIG_FILE): - print( - "First time use. Did not find credential file. Asking for " - "homeserver, user, and password to create credential file." - ) - homeserver = "https://matrix.example.org" - homeserver = input(f"Enter your homeserver URL: [{homeserver}] ") + # print( + # "First time use. Did not find credential file. Asking for " + # "homeserver, user, and password to create credential file." + # ) + # homeserver = "https://matrix.example.org" + # homeserver = input(f"Enter your homeserver URL: [{homeserver}] ") - if not (homeserver.startswith("https://") or homeserver.startswith("http://")): - homeserver = "https://" + homeserver + # if not (homeserver.startswith("https://") or homeserver.startswith("http://")): + # homeserver = "https://" + homeserver - user_id = "@user:example.org" - user_id = input(f"Enter your full user ID: [{user_id}] ") + # user_id = "@user:example.org" + # user_id = input(f"Enter your full user ID: [{user_id}] ") + + # device_name = "matrix-nio" + # device_name = input(f"Choose a name for this device: [{device_name}] ") + homeserver = "https://matrix.domnomnom.com" + user_id = "@getbot:matrix.domnomnom.com" + room_id = "!sZpfYzLsRbnIOKJlPH:matrix.domnomnom.com" + device_name = "whitebox-nio" - device_name = "matrix-nio" - device_name = input(f"Choose a name for this device: [{device_name}] ") if not os.path.exists(STORE_PATH): os.makedirs(STORE_PATH)