pypush-plus-plus/demo.py
2023-09-23 16:13:41 -04:00

191 lines
6.8 KiB
Python

import json
import logging
from base64 import b64decode, b64encode
from subprocess import PIPE, Popen
from rich.logging import RichHandler
import apns
import ids
import imessage
import trio
logging.basicConfig(
level=logging.NOTSET, format="%(message)s", datefmt="[%X]", handlers=[RichHandler()]
)
# Set sane log levels
logging.getLogger("urllib3").setLevel(logging.WARNING)
logging.getLogger("py.warnings").setLevel(logging.ERROR) # Ignore warnings from urllib3
logging.getLogger("asyncio").setLevel(logging.WARNING)
logging.getLogger("jelly").setLevel(logging.INFO)
logging.getLogger("nac").setLevel(logging.INFO)
logging.getLogger("apns").setLevel(logging.INFO)
logging.getLogger("albert").setLevel(logging.INFO)
logging.getLogger("ids").setLevel(logging.DEBUG)
logging.getLogger("bags").setLevel(logging.INFO)
logging.getLogger("imessage").setLevel(logging.INFO)
logging.captureWarnings(True)
process = Popen(["git", "rev-parse", "HEAD"], stdout=PIPE) # type: ignore
(commit_hash, err) = process.communicate()
exit_code = process.wait()
commit_hash = commit_hash.decode().strip()
# Try and load config.json
try:
with open("config.json", "r") as f:
CONFIG = json.load(f)
except FileNotFoundError:
CONFIG = {}
# Re-register if the commit hash has changed
if CONFIG.get("commit_hash") != commit_hash:
logging.warning("pypush commit is different, forcing re-registration...")
CONFIG["commit_hash"] = commit_hash
if "id" in CONFIG:
del CONFIG["id"]
def safe_b64decode(s):
try:
return b64decode(s)
except:
return None
async def main():
token = CONFIG.get("push", {}).get("token")
if token is not None:
token = b64decode(token)
else:
token = b""
push_creds = apns.PushCredentials(
CONFIG.get("push", {}).get("key", ""), CONFIG.get("push", {}).get("cert", ""), token)
async with apns.APNSConnection.start(push_creds) as conn:
await conn.set_state(1)
await conn.filter(["com.apple.madrid"])
user = ids.IDSUser(conn)
user.auth_and_set_encryption_from_config(CONFIG)
# Write config.json
CONFIG["encryption"] = {
"rsa_key": user.encryption_identity.encryption_key,
"ec_key": user.encryption_identity.signing_key,
}
CONFIG["id"] = {
"key": user._id_keypair.key,
"cert": user._id_keypair.cert,
}
CONFIG["auth"] = {
"key": user._auth_keypair.key,
"cert": user._auth_keypair.cert,
"user_id": user.user_id,
"handles": user.handles,
}
CONFIG["push"] = {
"token": b64encode(user.push_connection.credentials.token).decode(),
"key": user.push_connection.credentials.private_key,
"cert": user.push_connection.credentials.cert,
}
CONFIG["extra"] = user.extra
with open("config.json", "w") as f:
json.dump(CONFIG, f, indent=4)
im = imessage.iMessageUser(conn, user)
# Send a message to myself
async with trio.open_nursery() as nursery:
nursery.start_soon(input_task, im)
nursery.start_soon(output_task, im)
async def input_task(im: imessage.iMessageUser):
current_effect: str | None = None
current_participants: list[str] = []
def is_cmd(cmd_str: str, name: str) -> bool:
return cmd_str in [name, name[0]] or cmd_str.startswith(f"{name} ") or cmd_str.startswith(f"{name[0]} ")
def get_parameters(cmd: str, err_msg: str) -> list[str] | None:
sections: list[str] = cmd.split(" ")
if len(sections) < 2 or len(sections[1]) == 0:
print(err_msg)
return None
else:
return sections[1:]
def fixup_handle(handle: str) -> str:
if handle.startswith("tel:+") or handle.startswith("mailto:"):
return handle
elif handle.startswith("tel:"):
return "tel:+" + handle[4:]
elif handle.startswith("+"):
return "tel:" + handle
elif handle[0].isdigit():
# if the handle is 10 digits, assume it's a US number
if len(handle) == 10:
return "tel:+1" + handle
# otherwise just assume it's a full phone number
return "tel:+" + handle
else: # assume it's an email
return "mailto:" + handle
while True:
if (cmd := await trio.to_thread.run_sync(input, ">> ", cancellable=True)) == "":
continue
if is_cmd(cmd, "help"):
print("help (h): show this message")
print("quit (q): quit")
print("filter (f) [recipient]: set the current chat")
print("note: recipient must start with tel: or mailto: and include the country code")
print("effect (e): adds an iMessage effect to the next sent message")
print("handle [handle]: set the current handle (for sending messages with)")
print("\\: escape commands (will be removed from message)")
print("all other commands will be treated as message text to be sent")
elif is_cmd(cmd, "quit"):
exit(0)
elif is_cmd(cmd, "effect"):
if (effect := get_parameters(cmd, "effect [effect namespace]")) is not None:
print(f"next message will be sent with [{effect[0]}]")
current_effect = effect[0]
elif is_cmd(cmd, "filter"):
# set the current chat
if (participants := get_parameters(cmd, "filter [recipients]")) is not None:
fixed_participants: list[str] = list(map(fixup_handle, participants))
print(f"Filtering to {fixed_participants}")
current_participants = fixed_participants
elif is_cmd(cmd, "handle"):
handles: list[str] = im.user.handles
av_handles: str = "\n".join([f"\t{h}{' (current)' if h == im.user.current_handle else ''}" for h in handles])
err_str: str = f"handle [handle]\nAvailable handles:\n{av_handles}"
if (input_handles := get_parameters(cmd, err_str)) is not None:
handle = fixup_handle(input_handles[0])
if handle in handles:
print(f"Using {handle} as handle")
im.user.current_handle = handle
else:
print(f"Handle {handle} not found")
elif len(current_participants) > 0:
if cmd.startswith("\\"):
cmd = cmd[1:]
await im.send(imessage.iMessage.create(im, cmd, current_participants, current_effect))
current_effect = None
else:
print("No chat selected")
async def output_task(im: imessage.iMessageUser):
while True:
msg = await im.receive()
print(str(msg))
if __name__ == "__main__":
trio.run(main)