mirror of
https://github.com/Sneed-Group/pypush-plus-plus
synced 2024-12-23 11:22:42 -06:00
formatting and import organizing
This commit is contained in:
JJTech0130
parent
daf8868bb5
commit
b128ac514a
8 changed files with 250 additions and 154 deletions
11
albert.py
11
albert.py
|
|
@ -90,11 +90,14 @@ def generate_push_cert() -> tuple[str, str]:
|
|||
encoding=serialization.Encoding.PEM,
|
||||
format=serialization.PrivateFormat.TraditionalOpenSSL,
|
||||
encryption_algorithm=serialization.NoEncryption(),
|
||||
).decode("utf-8").strip(),
|
||||
protocol["device-activation"]["activation-record"]["DeviceCertificate"].decode(
|
||||
"utf-8"
|
||||
).strip(),
|
||||
)
|
||||
.decode("utf-8")
|
||||
.strip(),
|
||||
protocol["device-activation"]["activation-record"]["DeviceCertificate"]
|
||||
.decode("utf-8")
|
||||
.strip(),
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
private_key, cert = generate_push_cert()
|
||||
|
|
|
|||
53
apns.py
53
apns.py
|
|
@ -1,10 +1,13 @@
|
|||
from __future__ import annotations
|
||||
|
||||
import courier, albert
|
||||
from hashlib import sha1
|
||||
import random
|
||||
import threading
|
||||
import time
|
||||
import random
|
||||
from hashlib import sha1
|
||||
|
||||
import albert
|
||||
import courier
|
||||
|
||||
|
||||
class APNSConnection:
|
||||
incoming_queue = []
|
||||
|
|
@ -47,7 +50,9 @@ class APNSConnection:
|
|||
self.sock = courier.connect(self.private_key, self.cert)
|
||||
|
||||
# Start the queue filler thread
|
||||
self.queue_filler_thread = threading.Thread(target=self._queue_filler, daemon=True)
|
||||
self.queue_filler_thread = threading.Thread(
|
||||
target=self._queue_filler, daemon=True
|
||||
)
|
||||
self.queue_filler_thread.start()
|
||||
|
||||
def connect(self, root: bool = True, token: bytes = None):
|
||||
|
|
@ -56,15 +61,23 @@ class APNSConnection:
|
|||
flags |= 0b0100
|
||||
|
||||
if token is None:
|
||||
payload = _serialize_payload(7, [(2, 0x01.to_bytes()), (5, flags.to_bytes(4))])
|
||||
payload = _serialize_payload(
|
||||
7, [(2, 0x01.to_bytes()), (5, flags.to_bytes(4))]
|
||||
)
|
||||
else:
|
||||
payload = _serialize_payload(7, [(1, token), (2, 0x01.to_bytes()), (5, flags.to_bytes(4))])
|
||||
payload = _serialize_payload(
|
||||
7, [(1, token), (2, 0x01.to_bytes()), (5, flags.to_bytes(4))]
|
||||
)
|
||||
|
||||
self.sock.write(payload)
|
||||
|
||||
payload = self.wait_for_packet(8)
|
||||
|
||||
if payload == None or payload[0] != 8 or _get_field(payload[1], 1) != 0x00.to_bytes():
|
||||
if (
|
||||
payload == None
|
||||
or payload[0] != 8
|
||||
or _get_field(payload[1], 1) != 0x00.to_bytes()
|
||||
):
|
||||
raise Exception("Failed to connect")
|
||||
|
||||
self.token = _get_field(payload[1], 3)
|
||||
|
|
@ -85,24 +98,32 @@ class APNSConnection:
|
|||
if id is None:
|
||||
id = random.randbytes(4)
|
||||
|
||||
payload = _serialize_payload(0x0a,
|
||||
[(4, id),
|
||||
payload = _serialize_payload(
|
||||
0x0A,
|
||||
[
|
||||
(4, id),
|
||||
(1, sha1(topic.encode()).digest()),
|
||||
(2, self.token),
|
||||
(3, payload)])
|
||||
(3, payload),
|
||||
],
|
||||
)
|
||||
|
||||
self.sock.write(payload)
|
||||
|
||||
payload = self.wait_for_packet(0x0b)
|
||||
payload = self.wait_for_packet(0x0B)
|
||||
|
||||
if payload[1][0][1] != 0x00.to_bytes():
|
||||
raise Exception("Failed to send message")
|
||||
|
||||
def set_state(self, state: int):
|
||||
self.sock.write(_serialize_payload(0x14, [(1, state.to_bytes(1)), (2, 0x7FFFFFFF.to_bytes(4))]))
|
||||
self.sock.write(
|
||||
_serialize_payload(
|
||||
0x14, [(1, state.to_bytes(1)), (2, 0x7FFFFFFF.to_bytes(4))]
|
||||
)
|
||||
)
|
||||
|
||||
def keep_alive(self):
|
||||
self.sock.write(_serialize_payload(0x0c, []))
|
||||
self.sock.write(_serialize_payload(0x0C, []))
|
||||
|
||||
# TODO: Find a way to make this non-blocking
|
||||
# def expect_message(self) -> tuple[int, list[tuple[int, bytes]]] | None:
|
||||
|
|
@ -129,6 +150,7 @@ def _deserialize_field(stream: bytes) -> tuple[int, bytes]:
|
|||
value = stream[3 : 3 + length]
|
||||
return id, value
|
||||
|
||||
|
||||
# Note: Takes a stream, not a buffer, as we do not know the length of the payload
|
||||
# WILL BLOCK IF THE STREAM IS EMPTY
|
||||
def _deserialize_payload(stream) -> tuple[int, list[tuple[int, bytes]]] | None:
|
||||
|
|
@ -150,7 +172,10 @@ def _deserialize_payload(stream) -> tuple[int, list[tuple[int, bytes]]] | None:
|
|||
|
||||
return id, fields
|
||||
|
||||
def _deserialize_payload_from_buffer(buffer: bytes) -> tuple[int, list[tuple[int, bytes]]] | None:
|
||||
|
||||
def _deserialize_payload_from_buffer(
|
||||
buffer: bytes,
|
||||
) -> tuple[int, list[tuple[int, bytes]]] | None:
|
||||
id = int.from_bytes(buffer[:1], "big")
|
||||
|
||||
if id == 0x0:
|
||||
|
|
|
|||
|
|
@ -1,11 +1,12 @@
|
|||
import requests
|
||||
import plistlib
|
||||
|
||||
import requests
|
||||
|
||||
# CONFIG_URL = "http://init-p01st.push.apple.com/bag"
|
||||
CONFIG_URL = "https://init.push.apple.com/bag"
|
||||
|
||||
def get_config():
|
||||
|
||||
def get_config():
|
||||
r = requests.get(CONFIG_URL, verify=False)
|
||||
if r.status_code != 200:
|
||||
raise Exception("Failed to get config")
|
||||
|
|
@ -18,6 +19,7 @@ def get_config():
|
|||
|
||||
return config
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
config = get_config()
|
||||
print(config)
|
||||
|
|
@ -1,10 +1,12 @@
|
|||
import tlslite
|
||||
import socket
|
||||
|
||||
import tlslite
|
||||
|
||||
COURIER_HOST = "windows.courier.push.apple.com" # TODO: Get this from config
|
||||
COURIER_PORT = 5223
|
||||
ALPN = [b"apns-security-v2"]
|
||||
|
||||
|
||||
# Connect to the courier server
|
||||
def connect(private_key: str, cert: str) -> tlslite.TLSConnection:
|
||||
# Connect to the courier server
|
||||
|
|
@ -19,9 +21,9 @@ def connect(private_key: str, cert: str) -> tlslite.TLSConnection:
|
|||
|
||||
return sock
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sock = connect()
|
||||
sock.write(b"Hello World!")
|
||||
print(sock.read())
|
||||
sock.close()
|
||||
|
||||
|
|
|
|||
16
demo.py
16
demo.py
|
|
@ -1,17 +1,15 @@
|
|||
import plistlib
|
||||
import zlib
|
||||
from base64 import b64decode, b64encode
|
||||
from hashlib import sha1
|
||||
|
||||
import apns
|
||||
import ids
|
||||
|
||||
conn1 = apns.APNSConnection()
|
||||
conn1.connect()
|
||||
conn1.keep_alive()
|
||||
conn1.set_state(0x01)
|
||||
conn1.filter([])
|
||||
conn1.connect(False)
|
||||
|
||||
# Uncomment these for greater parity with apsd
|
||||
# conn1.keep_alive()
|
||||
# conn1.set_state(0x01)
|
||||
# conn1.filter([])
|
||||
# conn1.connect(False)
|
||||
|
||||
conn1.filter(["com.apple.madrid"])
|
||||
|
||||
print(ids.lookup(conn1, ["mailto:jjtech@jjtech.dev"]))
|
||||
|
|
|
|||
123
printer.py
123
printer.py
|
|
@ -1,6 +1,7 @@
|
|||
from base64 import b64encode, b64decode
|
||||
import plistlib
|
||||
import zlib
|
||||
from base64 import b64decode, b64encode
|
||||
from hashlib import sha1
|
||||
import plistlib, zlib
|
||||
|
||||
# Taken from debug logs of apsd
|
||||
enabled_topics = "(com.apple.icloud-container.com.apple.avatarsd, com.icloud.askpermission, com.apple.icloud-container.com.apple.Safari, com.apple.itunesstored, com.apple.icloud-container.clouddocs.com.apple.CloudDocs.health, com.apple.passd.usernotifications, com.apple.icloud-container.com.apple.donotdisturbd, com.apple.icloud-container.clouddocs.iCloud.com.reddit.reddit, com.apple.mobileme.fmf3, com.apple.icloud-container.com.apple.cloudpaird, com.apple.icloud-container.clouddocs.com.apple.Pages, com.apple.appstored-testflight, com.apple.askpermissiond, com.apple.icloud-container.com.apple.willowd, com.me.cal, com.apple.icloud-container.com.apple.suggestd, com.apple.icloud-container.clouddocs.F3LWYJ7GM7.com.apple.garageband10, com.apple.icloud-container.clouddocs.com.apple.CloudDocs.container-metadata, com.apple.icloud-container.com.apple.callhistory.sync-helper, com.apple.icloud-container.com.apple.syncdefaultsd, com.apple.icloud-container.com.apple.SafariShared.Settings, com.apple.pay.services.products.prod, com.apple.icloud-container.com.apple.StatusKitAgent, com.apple.icloud-container.com.apple.siriknowledged, com.me.contacts, com.apple.icloud-container.com.apple.TrustedPeersHelper, com.apple.icloud-container.clouddocs.iCloud.com.apple.iBooks, com.apple.icloud-container.clouddocs.iCloud.dk.simonbs.Scriptable, com.apple.icloud-container.clouddocs.com.apple.ScriptEditor2, com.icloud.family, com.apple.idmsauth, com.apple.watchList, com.apple.icloud-container.clouddocs.com.apple.TextEdit, com.apple.icloud-container.com.apple.VoiceMemos, com.apple.sharedstreams, com.apple.pay.services.apply.prod, com.apple.icloud-container.com.apple.SafariShared.CloudTabs, com.apple.wallet.sharing.qa, com.apple.appstored, com.apple.icloud-container.clouddocs.3L68KQB4HG.com.readdle.CommonDocuments, com.apple.icloud-container.clouddocs.com.apple.CloudDocs.pp-metadata, com.me.setupservice, com.apple.icloud-container.com.apple.amsengagementd, com.apple.icloud-container.com.apple.appleaccount.beneficiary.private, com.apple.icloud-container.com.apple.appleaccount.beneficiary, com.apple.icloud-container.clouddocs.com.apple.mail, com.apple.icloud-container.com.apple.appleaccount.custodian, com.apple.icloud-container.com.apple.securityd, com.apple.icloud-container.com.apple.iBooksX, com.apple.icloud-container.clouddocs.com.apple.QuickTimePlayerX, com.apple.icloud-container.clouddocs.com.apple.TextInput, com.apple.icloud-container.com.apple.icloud.fmfd, com.apple.tv.favoriteTeams, com.apple.pay.services.ownershipTokens.prod, com.apple.icloud-container.com.apple.passd, com.apple.amsaccountsd, com.apple.pay.services.devicecheckin.prod.us, com.apple.storekit, com.apple.icloud-container.com.apple.keyboardservicesd, paymentpass.com.apple, com.apple.aa.setupservice, com.apple.icloud-container.clouddocs.com.apple.shoebox, com.apple.icloud-container.clouddocs.F3LWYJ7GM7.com.apple.mobilegarageband, com.apple.icloud-container.com.apple.icloud.searchpartyuseragent, com.apple.icloud-container.clouddocs.iCloud.com.apple.configurator.ui, com.apple.icloud-container.com.apple.gamed, com.apple.icloud-container.clouddocs.com.apple.Keynote, com.apple.icloud-container.com.apple.willowd.homekit, com.apple.amsengagementd.notifications, com.apple.icloud.presence.mode.status, com.apple.aa.idms, com.apple.icloud-container.clouddocs.iCloud.com.apple.MobileSMS, com.apple.gamed, com.apple.icloud-container.clouddocs.iCloud.is.workflow.my.workflows, com.apple.icloud-container.clouddocs.iCloud.md.obsidian, com.apple.icloud-container.clouddocs.com.apple.CloudDocs, com.apple.wallet.sharing, com.apple.icloud-container.clouddocs.iCloud.com.apple.iBooks.iTunesU, com.apple.icloud.presence.shared.experience, com.apple.icloud-container.com.apple.imagent, com.apple.icloud-container.com.apple.financed, com.apple.pay.services.account.prod, com.apple.icloud-container.com.apple.assistant.assistantd, com.apple.pay.services.ck.zone.prod, com.apple.icloud-container.com.apple.security.cuttlefish, com.apple.icloud-container.clouddocs.com.apple.iBooks.cloudData, com.apple.peerpayment, com.icloud.quota, com.apple.pay.provision, com.apple.icloud-container.com.apple.upload-request-proxy.com.apple.photos.cloud, com.apple.icloud-container.com.apple.appleaccount.custodian.private, com.apple.icloud-container.clouddocs.com.apple.Preview, com.apple.maps.icloud, com.apple.icloud-container.com.apple.reminders, com.apple.icloud-container.com.apple.SafariShared.WBSCloudBookmarksStore, com.apple.idmsauthagent, com.apple.icloud-container.clouddocs.com.apple.Numbers, com.apple.bookassetd, com.apple.pay.auxiliary.registration.requirement.prod, com.apple.icloud.fmip.voiceassistantsync)"
|
||||
|
|
@ -17,16 +18,18 @@ topics = enabled_topics + opportunistic_topics + paused_topics
|
|||
# Calculate the SHA1 hash of each topic
|
||||
topics_lookup = [(topic, sha1(topic.encode()).digest()) for topic in topics]
|
||||
|
||||
|
||||
class bcolors:
|
||||
HEADER = '\033[95m'
|
||||
OKBLUE = '\033[94m'
|
||||
OKCYAN = '\033[96m'
|
||||
OKGREEN = '\033[92m'
|
||||
WARNING = '\033[93m'
|
||||
FAIL = '\033[91m'
|
||||
ENDC = '\033[0m'
|
||||
BOLD = '\033[1m'
|
||||
UNDERLINE = '\033[4m'
|
||||
HEADER = "\033[95m"
|
||||
OKBLUE = "\033[94m"
|
||||
OKCYAN = "\033[96m"
|
||||
OKGREEN = "\033[92m"
|
||||
WARNING = "\033[93m"
|
||||
FAIL = "\033[91m"
|
||||
ENDC = "\033[0m"
|
||||
BOLD = "\033[1m"
|
||||
UNDERLINE = "\033[4m"
|
||||
|
||||
|
||||
def _lookup_topic(hash: bytes):
|
||||
for topic_lookup in topics_lookup:
|
||||
|
|
@ -34,6 +37,7 @@ def _lookup_topic(hash: bytes):
|
|||
return topic_lookup[0]
|
||||
return None
|
||||
|
||||
|
||||
# Returns the value of the first field with the given id
|
||||
def _get_field(fields: list[tuple[int, bytes]], id: int) -> bytes:
|
||||
for field_id, value in fields:
|
||||
|
|
@ -41,6 +45,7 @@ def _get_field(fields: list[tuple[int, bytes]], id: int) -> bytes:
|
|||
return value
|
||||
return None
|
||||
|
||||
|
||||
def _p_filter(prefix, fields: list[tuple[int, bytes]]):
|
||||
enabled = []
|
||||
ignored = []
|
||||
|
|
@ -95,11 +100,17 @@ def _p_filter(prefix, fields: list[tuple[int, bytes]]):
|
|||
if len(paused) > 100:
|
||||
paused = paused[:100] + "..."
|
||||
# (Token: {token.decode()})
|
||||
print(f"{bcolors.OKGREEN}{prefix}{bcolors.ENDC}: {bcolors.OKCYAN}Filter{bcolors.ENDC} {bcolors.WARNING}Enabled{bcolors.ENDC}: {enabled} {bcolors.FAIL}Ignored{bcolors.ENDC}: {ignored} {bcolors.OKBLUE}Oppertunistic{bcolors.ENDC}: {oppertunistic} {bcolors.OKGREEN}Paused{bcolors.ENDC}: {paused}")
|
||||
print(
|
||||
f"{bcolors.OKGREEN}{prefix}{bcolors.ENDC}: {bcolors.OKCYAN}Filter{bcolors.ENDC} {bcolors.WARNING}Enabled{bcolors.ENDC}: {enabled} {bcolors.FAIL}Ignored{bcolors.ENDC}: {ignored} {bcolors.OKBLUE}Oppertunistic{bcolors.ENDC}: {oppertunistic} {bcolors.OKGREEN}Paused{bcolors.ENDC}: {paused}"
|
||||
)
|
||||
|
||||
|
||||
import apns
|
||||
|
||||
def pretty_print_payload(prefix, payload: tuple[int, list[tuple[int, bytes]]]) -> bytes | None:
|
||||
|
||||
def pretty_print_payload(
|
||||
prefix, payload: tuple[int, list[tuple[int, bytes]]]
|
||||
) -> bytes | None:
|
||||
id = payload[0]
|
||||
|
||||
if id == 9:
|
||||
|
|
@ -108,14 +119,25 @@ def pretty_print_payload(prefix, payload: tuple[int, list[tuple[int, bytes]]]) -
|
|||
token_str = ""
|
||||
if _get_field(payload[1], 3):
|
||||
token_str = f"{bcolors.WARNING}Token{bcolors.ENDC}: {b64encode(_get_field(payload[1], 3)).decode()}"
|
||||
print(f"{bcolors.OKGREEN}{prefix}{bcolors.ENDC}: {bcolors.OKCYAN}Connected{bcolors.ENDC} {token_str} {bcolors.OKBLUE}{_get_field(payload[1], 1).hex()}{bcolors.ENDC}")
|
||||
print(
|
||||
f"{bcolors.OKGREEN}{prefix}{bcolors.ENDC}: {bcolors.OKCYAN}Connected{bcolors.ENDC} {token_str} {bcolors.OKBLUE}{_get_field(payload[1], 1).hex()}{bcolors.ENDC}"
|
||||
)
|
||||
elif id == 7:
|
||||
print(f"{bcolors.OKGREEN}{prefix}{bcolors.ENDC}: {bcolors.OKCYAN}Connect Request{bcolors.ENDC}", end="")
|
||||
print(
|
||||
f"{bcolors.OKGREEN}{prefix}{bcolors.ENDC}: {bcolors.OKCYAN}Connect Request{bcolors.ENDC}",
|
||||
end="",
|
||||
)
|
||||
if _get_field(payload[1], 1):
|
||||
print(f" {bcolors.WARNING}Token{bcolors.ENDC}: {b64encode(_get_field(payload[1], 1)).decode()}", end="")
|
||||
if _get_field(payload[1], 0x0c):
|
||||
print(
|
||||
f" {bcolors.WARNING}Token{bcolors.ENDC}: {b64encode(_get_field(payload[1], 1)).decode()}",
|
||||
end="",
|
||||
)
|
||||
if _get_field(payload[1], 0x0C):
|
||||
print(f" {bcolors.OKBLUE}SIGNED{bcolors.ENDC}", end="")
|
||||
if _get_field(payload[1], 0x5) and int.from_bytes(_get_field(payload[1], 0x5)) & 0x4:
|
||||
if (
|
||||
_get_field(payload[1], 0x5)
|
||||
and int.from_bytes(_get_field(payload[1], 0x5)) & 0x4
|
||||
):
|
||||
print(f" {bcolors.FAIL}ROOT{bcolors.ENDC}", end="")
|
||||
print()
|
||||
|
||||
|
|
@ -140,22 +162,35 @@ def pretty_print_payload(prefix, payload: tuple[int, list[tuple[int, bytes]]]) -
|
|||
|
||||
out = apns._serialize_payload(payload[0], payload[1])
|
||||
# return out
|
||||
elif id == 0xc:
|
||||
print(f"{bcolors.OKGREEN}{prefix}{bcolors.ENDC}: {bcolors.OKCYAN}Keep Alive{bcolors.ENDC}")
|
||||
elif id == 0xd:
|
||||
print(f"{bcolors.OKGREEN}{prefix}{bcolors.ENDC}: {bcolors.OKCYAN}Keep Alive Ack{bcolors.ENDC}")
|
||||
elif id == 0xC:
|
||||
print(
|
||||
f"{bcolors.OKGREEN}{prefix}{bcolors.ENDC}: {bcolors.OKCYAN}Keep Alive{bcolors.ENDC}"
|
||||
)
|
||||
elif id == 0xD:
|
||||
print(
|
||||
f"{bcolors.OKGREEN}{prefix}{bcolors.ENDC}: {bcolors.OKCYAN}Keep Alive Ack{bcolors.ENDC}"
|
||||
)
|
||||
elif id == 0x14:
|
||||
print(f"{bcolors.OKGREEN}{prefix}{bcolors.ENDC}: {bcolors.OKCYAN}Set State{bcolors.ENDC}: {_get_field(payload[1], 1).hex()}")
|
||||
elif id == 0x1d or id == 0x20:
|
||||
print(f"{bcolors.OKGREEN}{prefix}{bcolors.ENDC}: {bcolors.WARNING}PubSub ??{bcolors.ENDC}")
|
||||
elif id == 0xe:
|
||||
print(f"{bcolors.OKGREEN}{prefix}{bcolors.ENDC}: {bcolors.WARNING}Token Confirmation{bcolors.ENDC}")
|
||||
elif id == 0xa:
|
||||
print(
|
||||
f"{bcolors.OKGREEN}{prefix}{bcolors.ENDC}: {bcolors.OKCYAN}Set State{bcolors.ENDC}: {_get_field(payload[1], 1).hex()}"
|
||||
)
|
||||
elif id == 0x1D or id == 0x20:
|
||||
print(
|
||||
f"{bcolors.OKGREEN}{prefix}{bcolors.ENDC}: {bcolors.WARNING}PubSub ??{bcolors.ENDC}"
|
||||
)
|
||||
elif id == 0xE:
|
||||
print(
|
||||
f"{bcolors.OKGREEN}{prefix}{bcolors.ENDC}: {bcolors.WARNING}Token Confirmation{bcolors.ENDC}"
|
||||
)
|
||||
elif id == 0xA:
|
||||
topic = ""
|
||||
# topic = _lookup_topic(_get_field(payload[1], 1))
|
||||
# if it has apsd -> APNs in the prefix, it's an outgoing notification
|
||||
if "apsd -> APNs" in prefix:
|
||||
print(f"{bcolors.OKGREEN}{prefix}{bcolors.ENDC}: {bcolors.OKBLUE}OUTGOING Notification{bcolors.ENDC}", end="")
|
||||
print(
|
||||
f"{bcolors.OKGREEN}{prefix}{bcolors.ENDC}: {bcolors.OKBLUE}OUTGOING Notification{bcolors.ENDC}",
|
||||
end="",
|
||||
)
|
||||
topic = _lookup_topic(_get_field(payload[1], 1))
|
||||
# topic = _lookup_topic(_get_field(payload[1], 1))
|
||||
# if b"bplist" in _get_field(payload[1], 3):
|
||||
|
|
@ -171,7 +206,10 @@ def pretty_print_payload(prefix, payload: tuple[int, list[tuple[int, bytes]]]) -
|
|||
# print(f" {bcolors.OKBLUE}{key}{bcolors.ENDC}: {plist[key]}", end="")
|
||||
|
||||
else:
|
||||
print(f"{bcolors.OKGREEN}{prefix}{bcolors.ENDC}: {bcolors.OKCYAN}Notification{bcolors.ENDC}", end="")
|
||||
print(
|
||||
f"{bcolors.OKGREEN}{prefix}{bcolors.ENDC}: {bcolors.OKCYAN}Notification{bcolors.ENDC}",
|
||||
end="",
|
||||
)
|
||||
topic = _lookup_topic(_get_field(payload[1], 2))
|
||||
# if b"bplist" in _get_field(payload[1], 3):
|
||||
# print(f" {bcolors.OKBLUE}Binary{bcolors.ENDC}", end="")
|
||||
|
|
@ -186,20 +224,28 @@ def pretty_print_payload(prefix, payload: tuple[int, list[tuple[int, bytes]]]) -
|
|||
if "cT" in payload:
|
||||
# It's HTTP over APNs
|
||||
if "hs" in payload:
|
||||
print(f" {bcolors.WARNING}HTTP Response{bcolors.ENDC}: {payload['hs']}", end="")
|
||||
print(
|
||||
f" {bcolors.WARNING}HTTP Response{bcolors.ENDC}: {payload['hs']}",
|
||||
end="",
|
||||
)
|
||||
else:
|
||||
print(f" {bcolors.WARNING}HTTP Request{bcolors.ENDC}", end="")
|
||||
# print(f" {bcolors.WARNING}HTTP{bcolors.ENDC} {payload['hs']}", end="")
|
||||
if "u" in payload:
|
||||
print(f" {bcolors.OKCYAN}URL{bcolors.ENDC}: {payload['u']}", end="")
|
||||
print(f" {bcolors.FAIL}Content Type{bcolors.ENDC}: {payload['cT']}", end="")
|
||||
print(
|
||||
f" {bcolors.FAIL}Content Type{bcolors.ENDC}: {payload['cT']}",
|
||||
end="",
|
||||
)
|
||||
if "h" in payload:
|
||||
print(f" {bcolors.FAIL}Headers{bcolors.ENDC}: {payload['h']}", end="")
|
||||
print(
|
||||
f" {bcolors.FAIL}Headers{bcolors.ENDC}: {payload['h']}", end=""
|
||||
)
|
||||
if "b" in payload:
|
||||
# What am I really supposed to put in WBITS? Got this from a random SO answer
|
||||
# print(payload["b"])
|
||||
body = zlib.decompress(payload['b'], 16+zlib.MAX_WBITS)
|
||||
if b'plist' in body:
|
||||
body = zlib.decompress(payload["b"], 16 + zlib.MAX_WBITS)
|
||||
if b"plist" in body:
|
||||
body = plistlib.loads(body)
|
||||
print(f" {bcolors.FAIL}Body{bcolors.ENDC}: {body}", end="")
|
||||
|
||||
|
|
@ -210,13 +256,16 @@ def pretty_print_payload(prefix, payload: tuple[int, list[tuple[int, bytes]]]) -
|
|||
# for field in payload[1]:
|
||||
# print(f"Field ID: {field[0]}")
|
||||
# print(f"Field Value: {field[1]}")
|
||||
elif id == 0xb:
|
||||
print(f"{bcolors.OKGREEN}{prefix}{bcolors.ENDC}: {bcolors.OKCYAN}Notification Ack{bcolors.ENDC} {bcolors.OKBLUE}{_get_field(payload[1], 8).hex()}{bcolors.ENDC}")
|
||||
elif id == 0xB:
|
||||
print(
|
||||
f"{bcolors.OKGREEN}{prefix}{bcolors.ENDC}: {bcolors.OKCYAN}Notification Ack{bcolors.ENDC} {bcolors.OKBLUE}{_get_field(payload[1], 8).hex()}{bcolors.ENDC}"
|
||||
)
|
||||
else:
|
||||
print(prefix, f"Payload ID: {hex(payload[0])}")
|
||||
for field in payload[1]:
|
||||
print(f"Field ID: {field[0]}")
|
||||
print(f"Field Value: {field[1]}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
print(f"{bcolors.OKGREEN}Enabled:{bcolors.ENDC}")
|
||||
|
|
@ -1,13 +1,13 @@
|
|||
# TLS server to proxy APNs traffic
|
||||
|
||||
import socket
|
||||
import tlslite
|
||||
import sys
|
||||
import threading
|
||||
|
||||
import sys
|
||||
import tlslite
|
||||
|
||||
# setting path
|
||||
sys.path.append('../')
|
||||
sys.path.append("../")
|
||||
|
||||
# APNs server to proxy traffic to
|
||||
APNS_HOST = "windows.courier.push.apple.com"
|
||||
|
|
@ -18,6 +18,7 @@ ALPN = b"apns-security-v3"
|
|||
|
||||
global_cnt = 0
|
||||
|
||||
|
||||
# Connect to the APNs server
|
||||
def connect() -> tlslite.TLSConnection:
|
||||
# Connect to the APNs server
|
||||
|
|
@ -31,6 +32,7 @@ def connect() -> tlslite.TLSConnection:
|
|||
ssock.handshakeClientCert(alpn=[ALPN])
|
||||
else:
|
||||
import albert
|
||||
|
||||
private_key, cert = albert.generate_push_cert()
|
||||
cert = tlslite.X509CertChain([tlslite.X509().parse(cert)])
|
||||
private_key = tlslite.parsePEMKey(private_key, private=True)
|
||||
|
|
@ -39,6 +41,7 @@ def connect() -> tlslite.TLSConnection:
|
|||
|
||||
return ssock
|
||||
|
||||
|
||||
cert: str = None
|
||||
key: str = None
|
||||
|
||||
|
|
@ -50,6 +53,7 @@ outgoing_list = []
|
|||
incoming_list = []
|
||||
# last_outgoing = b""
|
||||
|
||||
|
||||
def proxy(conn1: tlslite.TLSConnection, conn2: tlslite.TLSConnection, prefix: str = ""):
|
||||
try:
|
||||
while True:
|
||||
|
|
@ -62,13 +66,17 @@ def proxy(conn1: tlslite.TLSConnection, conn2: tlslite.TLSConnection, prefix: st
|
|||
break
|
||||
|
||||
try:
|
||||
override = printer.pretty_print_payload(prefix, apns._deserialize_payload_from_buffer(data))
|
||||
override = printer.pretty_print_payload(
|
||||
prefix, apns._deserialize_payload_from_buffer(data)
|
||||
)
|
||||
except Exception as e:
|
||||
print(e) # Can't crash the proxy over parsing errors
|
||||
if override is not None:
|
||||
data = override
|
||||
print("OVERRIDE: ", end="")
|
||||
printer.pretty_print_payload(prefix, apns._deserialize_payload_from_buffer(data))
|
||||
printer.pretty_print_payload(
|
||||
prefix, apns._deserialize_payload_from_buffer(data)
|
||||
)
|
||||
|
||||
if "apsd -> APNs" in prefix:
|
||||
global outgoing_list
|
||||
|
|
@ -97,7 +105,10 @@ def proxy(conn1: tlslite.TLSConnection, conn2: tlslite.TLSConnection, prefix: st
|
|||
conn1.close()
|
||||
conn2.close()
|
||||
|
||||
|
||||
repl_lock = False
|
||||
|
||||
|
||||
def repl(conn1: tlslite.TLSConnection, conn2: tlslite.TLSConnection):
|
||||
global repl_lock
|
||||
if repl_lock:
|
||||
|
|
@ -105,8 +116,10 @@ def repl(conn1: tlslite.TLSConnection, conn2: tlslite.TLSConnection):
|
|||
return
|
||||
repl_lock = True
|
||||
import IPython
|
||||
|
||||
IPython.embed()
|
||||
|
||||
|
||||
def handle(conn: socket.socket):
|
||||
# Wrap the socket in TLS
|
||||
s_conn = tlslite.TLSConnection(conn)
|
||||
|
|
@ -117,7 +130,9 @@ def handle(conn: socket.socket):
|
|||
# cert = tlslite.X509CertChain([tlslite.X509().parse(cert)])
|
||||
key_parsed = tlslite.parsePEMKey(key, private=True)
|
||||
# print(key_parsed)
|
||||
s_conn.handshakeServer(certChain=chain, privateKey=key_parsed, reqCert=False, alpn=[ALPN])
|
||||
s_conn.handshakeServer(
|
||||
certChain=chain, privateKey=key_parsed, reqCert=False, alpn=[ALPN]
|
||||
)
|
||||
|
||||
print("Handling connection")
|
||||
# Connect to the APNs server
|
||||
|
|
@ -130,12 +145,14 @@ def handle(conn: socket.socket):
|
|||
global_cnt += 1
|
||||
# Proxy data between the connections
|
||||
# Create a thread to proxy data from the APNs server to the client
|
||||
threading.Thread(target=proxy, args=(s_conn, apns, f"{global_cnt} apsd -> APNs")).start()
|
||||
threading.Thread(
|
||||
target=proxy, args=(s_conn, apns, f"{global_cnt} apsd -> APNs")
|
||||
).start()
|
||||
# Just proxy data from the client to the APNs server in this thread
|
||||
proxy(apns, s_conn, f"{global_cnt} APNs -> apsd")
|
||||
|
||||
def serve():
|
||||
|
||||
def serve():
|
||||
# Create a socket to listen for connections
|
||||
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
||||
# Allow the socket to be reused
|
||||
|
|
@ -151,7 +168,6 @@ def serve():
|
|||
global cert
|
||||
cert = f.read()
|
||||
|
||||
|
||||
# NEED TO USE OPENSSL, SEE CORETRUST CMD, MIMIC ENTRUST? OR AT LEAST SEE PUSHPROXY FOR EXTRACTION & REPLACEMENT
|
||||
with open("push_key.pem", "r") as f:
|
||||
global key
|
||||
|
|
@ -174,5 +190,6 @@ def serve():
|
|||
conn.close()
|
||||
sock.close()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
serve()
|
||||
|
|
|
|||
Loading…
Reference in a new issue