Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
#!/usr/bin/python3
import asyncio
import getpass
import sys
import aioxmpp
async def amain(address, password, override_peer, no_verify, pubsub_service, pubsub_node):
client = aioxmpp.PresenceManagedClient(
address,
aioxmpp.make_security_layer(password, no_verify=no_verify),
override_peer=override_peer,
)
disco = client.summon(aioxmpp.DiscoClient)
pubsub = client.summon(aioxmpp.PubSubClient)
async with client.connected() as stream:
print("connected")
await pubsub.purge(pubsub_service, pubsub_node)
def jid(s):
return aioxmpp.JID.fromstr(s)
def main():
import argparse
parser = argparse.ArgumentParser()
parser.add_argument(
"address",
help="The JID/address to log in with",
type=jid,
)
parser.add_argument(
"--host", "-H",
default=None,
help="Override the host to connect to",
)
parser.add_argument(
"--port", "-p",
type=int,
default=5222,
help="If --host is given, override the port, too (default: 5222)"
)
parser.add_argument(
"--no-verify",
action="store_true",
help="Disable TLS verification (don’t)",
default=False,
)
parser.add_argument(
"pubsub_service",
help="Address of the pubsub service",
type=jid,
)
parser.add_argument(
"pubsub_node",
help="Name of the pubsub node to purge",
)
args = parser.parse_args()
override_peer = []
if args.host is not None:
override_peer.append((args.host, args.port, aioxmpp.connector.STARTTLSConnector()))
password = getpass.getpass()
loop = asyncio.get_event_loop()
try:
loop.run_until_complete(amain(
args.address, password,
override_peer,
args.no_verify,
args.pubsub_service,
args.pubsub_node,
))
finally:
loop.close()
if __name__ == "__main__":
sys.exit(main() or 0)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment