mirror of
https://github.com/Sneed-Group/pypush-plus-plus
synced 2024-12-23 19:32:29 -06:00
add basic tui
This commit is contained in:
JJTech0130
parent
565dfb32ec
commit
9838294cf0
2 changed files with 65 additions and 79 deletions
88
demo.py
88
demo.py
|
|
@ -19,10 +19,11 @@ logging.basicConfig(
|
||||||
logging.getLogger("urllib3").setLevel(logging.WARNING)
|
logging.getLogger("urllib3").setLevel(logging.WARNING)
|
||||||
logging.getLogger("jelly").setLevel(logging.INFO)
|
logging.getLogger("jelly").setLevel(logging.INFO)
|
||||||
logging.getLogger("nac").setLevel(logging.INFO)
|
logging.getLogger("nac").setLevel(logging.INFO)
|
||||||
logging.getLogger("apns").setLevel(logging.DEBUG)
|
logging.getLogger("apns").setLevel(logging.INFO)
|
||||||
logging.getLogger("albert").setLevel(logging.INFO)
|
logging.getLogger("albert").setLevel(logging.INFO)
|
||||||
logging.getLogger("ids").setLevel(logging.DEBUG)
|
logging.getLogger("ids").setLevel(logging.DEBUG)
|
||||||
logging.getLogger("bags").setLevel(logging.DEBUG)
|
logging.getLogger("bags").setLevel(logging.INFO)
|
||||||
|
logging.getLogger("imessage").setLevel(logging.DEBUG)
|
||||||
|
|
||||||
# Try and load config.json
|
# Try and load config.json
|
||||||
try:
|
try:
|
||||||
|
|
@ -108,61 +109,42 @@ with open("config.json", "w") as f:
|
||||||
|
|
||||||
im = imessage.iMessageUser(conn, user)
|
im = imessage.iMessageUser(conn, user)
|
||||||
|
|
||||||
# while True:
|
INPUT_QUEUE = apns.IncomingQueue()
|
||||||
# i = im.receive()
|
|
||||||
# if i is not None:
|
|
||||||
# print(f"Got message {i}")
|
|
||||||
imsg = imessage.iMessage()
|
|
||||||
imsg.sender = user.handles[0]
|
|
||||||
imsg.participants = ["mailto:jjtech@jjtech.dev"]
|
|
||||||
imsg.text = "Hello world!"
|
|
||||||
|
|
||||||
time.sleep(1)
|
def input_thread():
|
||||||
im.send(imsg)
|
while True:
|
||||||
|
from prompt_toolkit import prompt
|
||||||
|
|
||||||
for i in range(10):
|
try:
|
||||||
z = im.receive()
|
msg = prompt('>> ')
|
||||||
if z is not None:
|
except:
|
||||||
print(f"Got message {z}")
|
msg = 'quit'
|
||||||
time.sleep(1)
|
INPUT_QUEUE.append(msg)
|
||||||
# # Create a thread to take user input
|
|
||||||
# INPUT_QUEUE = apns.IncomingQueue()
|
|
||||||
|
|
||||||
# def input_thread():
|
threading.Thread(target=input_thread, daemon=True).start()
|
||||||
# while True:
|
|
||||||
# from prompt_toolkit import prompt
|
|
||||||
|
|
||||||
# try:
|
|
||||||
# msg = prompt('>> ')
|
|
||||||
# except:
|
|
||||||
# msg = 'quit'
|
|
||||||
# INPUT_QUEUE.append(msg)
|
|
||||||
|
|
||||||
# threading.Thread(target=input_thread, daemon=True).start()
|
|
||||||
|
|
||||||
|
|
||||||
# while True:
|
while True:
|
||||||
# msg = im.receive()
|
msg = im.receive()
|
||||||
# if msg is not None:
|
if msg is not None:
|
||||||
# print(f"Got message {msg}")
|
print(f"Got message {msg}")
|
||||||
|
|
||||||
# if len(INPUT_QUEUE) > 0:
|
if len(INPUT_QUEUE) > 0:
|
||||||
# msg = INPUT_QUEUE.pop()
|
msg = INPUT_QUEUE.pop()
|
||||||
# if msg == 'help' or msg == 'h':
|
if msg == 'help' or msg == 'h':
|
||||||
# print('help (h): show this message')
|
print('help (h): show this message')
|
||||||
# print('quit (q): quit')
|
print('quit (q): quit')
|
||||||
# print('send (s) [recipiant] [message]: send a message')
|
print('send (s) [recipiant] [message]: send a message')
|
||||||
# elif msg == 'quit' or msg == 'q':
|
elif msg == 'quit' or msg == 'q':
|
||||||
# break
|
break
|
||||||
# elif msg.startswith('send') or msg.startswith('s'):
|
elif msg.startswith('send') or msg.startswith('s'):
|
||||||
# msg = msg.split(' ')
|
msg = msg.split(' ')
|
||||||
# if len(msg) < 3:
|
if len(msg) < 3:
|
||||||
# print('send [recipiant] [message]')
|
print('send [recipiant] [message]')
|
||||||
# else:
|
else:
|
||||||
# imsg = imessage.iMessage()
|
im.send(imessage.iMessage(
|
||||||
# imsg.text = ' '.join(msg[2:])
|
text=' '.join(msg[2:]),
|
||||||
# imsg.participants = [msg[1], user.handles[0]]
|
participants=[msg[1], user.handles[0]],
|
||||||
# imsg.sender = user.handles[0]
|
#sender=user.handles[0]
|
||||||
|
))
|
||||||
# im.send(imsg)
|
|
||||||
|
|
||||||
|
|
|
||||||
56
imessage.py
56
imessage.py
|
|
@ -179,7 +179,7 @@ class iMessageUser:
|
||||||
payload = BytesIO(payload)
|
payload = BytesIO(payload)
|
||||||
|
|
||||||
tag = payload.read(1)
|
tag = payload.read(1)
|
||||||
print("TAG", tag)
|
#print("TAG", tag)
|
||||||
body_length = int.from_bytes(payload.read(2), "big")
|
body_length = int.from_bytes(payload.read(2), "big")
|
||||||
body = payload.read(body_length)
|
body = payload.read(body_length)
|
||||||
|
|
||||||
|
|
@ -289,7 +289,7 @@ class iMessageUser:
|
||||||
|
|
||||||
cipher = Cipher(algorithms.AES(rsa_body[:16]), modes.CTR(NORMAL_NONCE))
|
cipher = Cipher(algorithms.AES(rsa_body[:16]), modes.CTR(NORMAL_NONCE))
|
||||||
decrypted = cipher.decryptor().update(rsa_body[16:] + body.read())
|
decrypted = cipher.decryptor().update(rsa_body[16:] + body.read())
|
||||||
|
|
||||||
return decrypted
|
return decrypted
|
||||||
|
|
||||||
def _verify_payload(self, payload: bytes, sender: str, sender_token: str) -> bool:
|
def _verify_payload(self, payload: bytes, sender: str, sender_token: str) -> bool:
|
||||||
|
|
@ -325,7 +325,7 @@ class iMessageUser:
|
||||||
return None
|
return None
|
||||||
body = apns._get_field(raw[1], 3)
|
body = apns._get_field(raw[1], 3)
|
||||||
body = plistlib.loads(body)
|
body = plistlib.loads(body)
|
||||||
print(f"Got body message {body}")
|
#print(f"Got body message {body}")
|
||||||
payload = body["P"]
|
payload = body["P"]
|
||||||
|
|
||||||
if not self._verify_payload(payload, body['sP'], body["t"]):
|
if not self._verify_payload(payload, body['sP'], body["t"]):
|
||||||
|
|
@ -419,28 +419,32 @@ class iMessageUser:
|
||||||
|
|
||||||
self.connection.send_message("com.apple.madrid", body, msg_id)
|
self.connection.send_message("com.apple.madrid", body, msg_id)
|
||||||
|
|
||||||
def check_response(x):
|
# This code can check to make sure we got a success response, but waiting for the response is annoying,
|
||||||
if x[0] != 0x0A:
|
# so for now we just YOLO it and assume it worked
|
||||||
return False
|
|
||||||
if apns._get_field(x[1], 2) != sha1("com.apple.madrid".encode()).digest():
|
|
||||||
return False
|
|
||||||
resp_body = apns._get_field(x[1], 3)
|
|
||||||
if resp_body is None:
|
|
||||||
return False
|
|
||||||
resp_body = plistlib.loads(resp_body)
|
|
||||||
if "c" not in resp_body or resp_body["c"] != 255:
|
|
||||||
return False
|
|
||||||
return True
|
|
||||||
|
|
||||||
num_recv = 0
|
# def check_response(x):
|
||||||
while True:
|
# if x[0] != 0x0A:
|
||||||
if num_recv == len(bundled_payloads):
|
# return False
|
||||||
break
|
# if apns._get_field(x[1], 2) != sha1("com.apple.madrid".encode()).digest():
|
||||||
payload = self.connection.incoming_queue.wait_pop_find(check_response)
|
# return False
|
||||||
if payload is None:
|
# resp_body = apns._get_field(x[1], 3)
|
||||||
continue
|
# if resp_body is None:
|
||||||
|
# return False
|
||||||
|
# resp_body = plistlib.loads(resp_body)
|
||||||
|
# if "c" not in resp_body or resp_body["c"] != 255:
|
||||||
|
# return False
|
||||||
|
# return True
|
||||||
|
|
||||||
|
|
||||||
resp_body = apns._get_field(payload[1], 3)
|
# num_recv = 0
|
||||||
resp_body = plistlib.loads(resp_body)
|
# while True:
|
||||||
logger.error(resp_body)
|
# if num_recv == len(bundled_payloads):
|
||||||
num_recv += 1
|
# break
|
||||||
|
# payload = self.connection.incoming_queue.wait_pop_find(check_response)
|
||||||
|
# if payload is None:
|
||||||
|
# continue
|
||||||
|
|
||||||
|
# resp_body = apns._get_field(payload[1], 3)
|
||||||
|
# resp_body = plistlib.loads(resp_body)
|
||||||
|
# logger.error(resp_body)
|
||||||
|
# num_recv += 1
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue