Sneed-Group/pypush-plus-plus
1
0
Fork 0
mirror of https://github.com/Sneed-Group/pypush-plus-plus synced 2024-12-23 19:32:29 -06:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge pull request #47 from SpaceSaver/sms-registration

Browse source
This commit is contained in:
JJTech 2023-11-17 23:28:05 +01:00 committed by GitHub
commit 7dd62e93c9
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
1 changed files with 33 additions and 15 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

48
demo.py
View file

@ -81,6 +81,23 @@ async def main(args: argparse.Namespace):
users = ids.register(conn, users, vd, args.client_data or args.reg_notify)
return users
def expiration_identifier(users: list[ids.IDSUser]) -> datetime.datetime | None:
expiration = None
# Format time as HH:MM:SS PM/AM EST/EDT (X minutes from now)
expire_msg = lambda expiration: f"Number registration is valid until {str(expiration.astimezone().strftime('%x %I:%M:%S %p %Z'))} ({str(int((expiration - datetime.datetime.now(datetime.timezone.utc)).total_seconds()/60))} minutes from now)"
for user in users:
# If this is a phone number user, then it's got to be the one we just linked
# so pull out the expiration date from the certificate
if "P:" in str(user.user_id):
# There is not really a good reason to try/catch here: If we couldn't reregister, just crash (very unlikely we can recover)
cert = x509.load_pem_x509_certificate(user.id_cert.encode('utf-8'))
expiration = cert.not_valid_after
# Make it a UTC aware timezone, for reasons
expiration = expiration.replace(tzinfo=datetime.timezone.utc)
logging.info(expire_msg(expiration))
return expiration
async def reregister(conn: apns.APNSConnection, users: list[ids.IDSUser]) -> datetime.datetime:
register(conn, users)
@ -89,6 +106,7 @@ async def main(args: argparse.Namespace):
expiration = None
# Format time as HH:MM:SS PM/AM EST/EDT (X minutes from now)
expire_msg = lambda expiration: f"Number registration is valid until {str(expiration.astimezone().strftime('%x %I:%M:%S %p %Z'))} ({str(int((expiration - datetime.datetime.now(datetime.timezone.utc)).total_seconds()/60))} minutes from now)"
email_user = None
email_addr = None # For HACK below
@ -104,17 +122,7 @@ async def main(args: argparse.Namespace):
"handles": user.handles,
})
# If this is a phone number user, then it's got the be the one we just linked
# so pull out the expiration date from the certificate
if "P:" in str(user.user_id):
# There is not really a good reason to try/catch here: If we couldn't reregister, just crash (very unlikely we can recover)
cert = x509.load_pem_x509_certificate(user.id_cert.encode('utf-8'))
expiration = cert.not_valid_after
# Make it a UTC aware timezone, for reasons
expiration = expiration.replace(tzinfo=datetime.timezone.utc)
logging.info(expire_msg(expiration))
else:
if not "P:" in str(user.user_id):
email_user = user
for n in range(len(user.handles)):
# HACK: Just pick the first email address they have to avoid picking the linked phone number
@ -122,6 +130,8 @@ async def main(args: argparse.Namespace):
if "mailto:" in str(user.handles[n]):
email_addr = user.handles[n]
expiration = expiration_identifier(users)
# Save the config to disk
safe_config()
@ -206,22 +216,30 @@ async def main(args: argparse.Namespace):
if args.daemon:
wait_time_minutes = 5 # this is in minutes. 5 recommended
expiration = await reregister(conn, users)
if args.reregister:
expiration = await reregister(conn, users)
else:
expiration = expiration_identifier(users)
if expiration is None:
expiration = await reregister(conn, users)
while True:
reregister_time = expiration - datetime.timedelta(minutes=wait_time_minutes) # wait_time_minutes before expiration
reregister_delta = (reregister_time - datetime.datetime.now(datetime.timezone.utc)).total_seconds()
logging.info(f"Reregistering in {int(reregister_delta / 60)} minutes...")
await trio.sleep(reregister_delta)
if (reregister_delta > 0):
await trio.sleep(reregister_delta)
logging.info("Reregistering...")
expiration = await reregister(conn, users)
logging.info("Reregistered!")
if args.reregister:
elif args.reregister:
await reregister(conn, users)
print("Done!")