more click, more api, even more fixes
This commit is contained in:
parent
8fcf9b4785
commit
cc8c77e780
18 changed files with 318 additions and 189 deletions
1
smal/.gitignore
vendored
1
smal/.gitignore
vendored
|
|
@ -1,2 +1,3 @@
|
|||
__pycache__
|
||||
*.egg-info
|
||||
build
|
||||
|
|
|
|||
|
|
@ -0,0 +1 @@
|
|||
from .demobot import main
|
||||
|
|
@ -0,0 +1 @@
|
|||
from .demobot import main
|
||||
|
|
@ -1,9 +1,11 @@
|
|||
# Copyright (C) 2026 saces@c-base.org
|
||||
# SPDX-License-Identifier: AGPL-3.0-only
|
||||
import sys
|
||||
from _pygomx import lib, ffi
|
||||
import click
|
||||
import json
|
||||
from pygomx.errors import PygomxAPIError
|
||||
|
||||
from pygomx import CliV0
|
||||
|
||||
from .click import click_catch_exception
|
||||
|
||||
|
||||
@click.command()
|
||||
|
|
@ -11,18 +13,12 @@ import json
|
|||
"--json", "show_json", is_flag=True, help="show json as returned from server."
|
||||
)
|
||||
@click.argument("domain", metavar="string")
|
||||
@click_catch_exception(handle=(PygomxAPIError))
|
||||
def discoverhs(domain, show_json):
|
||||
"""Attempts to discover the homeserver from the given string"""
|
||||
mxid = domain.encode(encoding="utf-8")
|
||||
result = CliV0.Discover(domain)
|
||||
|
||||
r = lib.cliv0_discoverhs(mxid)
|
||||
result = ffi.string(r).decode("utf-8")
|
||||
lib.FreeCString(r)
|
||||
if result.startswith("ERR:"):
|
||||
print(result)
|
||||
sys.exit(1)
|
||||
if show_json:
|
||||
print(result)
|
||||
click.echo(result)
|
||||
else:
|
||||
result_dict = json.loads(result)
|
||||
print(result_dict["m.homeserver"]["base_url"])
|
||||
click.echo(result["m.homeserver"]["base_url"])
|
||||
|
|
|
|||
|
|
@ -4,9 +4,12 @@ import getpass
|
|||
from datetime import datetime
|
||||
|
||||
import click
|
||||
from pygomx.errors import PygomxAPIError
|
||||
|
||||
from pygomx import CliV0
|
||||
|
||||
from .click import click_catch_exception
|
||||
|
||||
|
||||
@click.command()
|
||||
@click.option("-u", "--url", "hs_url", metavar="url", help="homeserver url")
|
||||
|
|
@ -22,17 +25,18 @@ from pygomx import CliV0
|
|||
type=click.Choice(["all", "other", "self"]),
|
||||
help="logout devices",
|
||||
)
|
||||
@click_catch_exception(handle=(PygomxAPIError))
|
||||
def logout(hs_url, token, devices, logout_type, show_json):
|
||||
"""List or logout devices.
|
||||
|
||||
\b
|
||||
mxlogout [--json]
|
||||
list all devices
|
||||
mxlogout --all
|
||||
mxlogout --logout all
|
||||
logout all devices
|
||||
mxlogout --self
|
||||
mxlogout --logout self
|
||||
logout this device
|
||||
mxlogout --other
|
||||
mxlogout --logout other
|
||||
logout all other devices (requires auth)
|
||||
mxlogout deviceid [deviceid]...
|
||||
logout given devices (requires auth)
|
||||
|
|
@ -79,7 +83,7 @@ def logout(hs_url, token, devices, logout_type, show_json):
|
|||
return
|
||||
|
||||
if show_json:
|
||||
print(raw_device_dict)
|
||||
click.echo(raw_device_dict)
|
||||
return
|
||||
|
||||
max_len = 0
|
||||
|
|
@ -88,12 +92,16 @@ def logout(hs_url, token, devices, logout_type, show_json):
|
|||
|
||||
for device in raw_device_dict["devices"]:
|
||||
date_object = datetime.fromtimestamp(device["last_seen_ts"] / 1000)
|
||||
print(
|
||||
device["device_id"],
|
||||
" " * (max_len - len(device["device_id"])),
|
||||
date_object,
|
||||
device["last_seen_ip"],
|
||||
device["display_name"],
|
||||
click.echo(
|
||||
" ".join(
|
||||
[
|
||||
device["device_id"],
|
||||
" " * (max_len - len(device["device_id"])),
|
||||
str(date_object),
|
||||
device["last_seen_ip"],
|
||||
device["display_name"],
|
||||
]
|
||||
)
|
||||
)
|
||||
date_object = datetime.fromtimestamp(device["last_seen_ts"] / 1000)
|
||||
|
||||
|
|
@ -106,7 +114,7 @@ def do_logout(cli, all):
|
|||
if all:
|
||||
reqData["path"] += ["all"]
|
||||
res = cli.Generic(reqData)
|
||||
print(res)
|
||||
click.echo(res)
|
||||
|
||||
|
||||
def do_logout_devices(cli, devices, user_id):
|
||||
|
|
@ -126,4 +134,4 @@ def do_logout_devices(cli, devices, user_id):
|
|||
},
|
||||
}
|
||||
res = cli.Generic(reqData)
|
||||
print(res)
|
||||
click.echo(res)
|
||||
|
|
|
|||
|
|
@ -1,9 +1,14 @@
|
|||
# Copyright (C) 2026 saces@c-base.org
|
||||
# SPDX-License-Identifier: AGPL-3.0-only
|
||||
from _pygomx import lib, ffi
|
||||
import click
|
||||
import json
|
||||
|
||||
import click
|
||||
from pygomx.errors import PygomxAPIError
|
||||
|
||||
from pygomx import CliV0
|
||||
|
||||
from .click import click_catch_exception
|
||||
|
||||
|
||||
@click.command()
|
||||
@click.option(
|
||||
|
|
@ -15,6 +20,7 @@ import json
|
|||
)
|
||||
@click.option("-d", "--domain", "domain", metavar="domain", help="domain selector")
|
||||
@click.argument("mxpassfile", metavar="mxpassfilepath", required=False)
|
||||
@click_catch_exception(handle=(PygomxAPIError))
|
||||
def passitem(mxpassfile, show_secret, hs_url, localpart, domain):
|
||||
"""utility to get items from mxpasss files"""
|
||||
|
||||
|
|
@ -28,19 +34,10 @@ def passitem(mxpassfile, show_secret, hs_url, localpart, domain):
|
|||
if domain is None:
|
||||
domain = "*"
|
||||
|
||||
r = lib.cliv0_mxpassitem(
|
||||
mxpassfile.encode(encoding="utf-8"),
|
||||
hs_url.encode(encoding="utf-8"),
|
||||
localpart.encode(encoding="utf-8"),
|
||||
domain.encode(encoding="utf-8"),
|
||||
)
|
||||
result = ffi.string(r).decode("utf-8")
|
||||
lib.FreeCString(r)
|
||||
|
||||
result_dict = json.loads(result)
|
||||
result_dict = CliV0.MXPassItem(mxpassfile, hs_url, localpart, domain)
|
||||
|
||||
if show_secret:
|
||||
print(result_dict["Token"])
|
||||
click.echo(result_dict["Token"])
|
||||
else:
|
||||
result_dict["Token"] = "***"
|
||||
print(json.dumps(result_dict))
|
||||
click.echo(json.dumps(result_dict))
|
||||
|
|
|
|||
|
|
@ -2,7 +2,7 @@
|
|||
# SPDX-License-Identifier: AGPL-3.0-only
|
||||
import click
|
||||
from pygomx import CliV0
|
||||
import pygomx
|
||||
from pygomx.errors import PygomxAPIError
|
||||
|
||||
from .click import click_catch_exception
|
||||
|
||||
|
|
@ -10,7 +10,7 @@ from .click import click_catch_exception
|
|||
@click.command()
|
||||
@click.option("-u", "--url", "hs_url", metavar="url", help="homeserver url")
|
||||
@click.option("-t", "--token", "token", metavar="token", help="access token")
|
||||
@click_catch_exception(handle=(pygomx.errors.APIError))
|
||||
@click_catch_exception(handle=(PygomxAPIError))
|
||||
def whoami(hs_url, token):
|
||||
"""this token belongs to?"""
|
||||
|
||||
|
|
|
|||
|
|
@ -1,49 +1,60 @@
|
|||
# Copyright (C) 2026 saces@c-base.org
|
||||
# SPDX-License-Identifier: AGPL-3.0-only
|
||||
import sys
|
||||
import os
|
||||
import datetime
|
||||
import getpass
|
||||
import json
|
||||
from _pygomx import lib, ffi
|
||||
import os
|
||||
import time
|
||||
from functools import partial, wraps
|
||||
|
||||
import click
|
||||
from pygomx.errors import PygomxAPIError
|
||||
|
||||
from pygomx import ApiV0
|
||||
|
||||
|
||||
def smalsetup():
|
||||
if len(sys.argv) != 2:
|
||||
print("usage: ", sys.argv[0], " matrixid")
|
||||
return 1
|
||||
def catch_exception(func=None, *, handle):
|
||||
if not func:
|
||||
return partial(catch_exception, handle=handle)
|
||||
|
||||
mxid = sys.argv[1].encode(encoding="utf-8")
|
||||
@wraps(func)
|
||||
def wrapper(*args, **kwargs):
|
||||
try:
|
||||
return func(*args, **kwargs)
|
||||
except handle as e:
|
||||
raise click.ClickException(e)
|
||||
|
||||
r = lib.apiv0_discover(mxid)
|
||||
result = ffi.string(r).decode("utf-8")
|
||||
lib.FreeCString(r)
|
||||
return wrapper
|
||||
|
||||
if result.startswith("ERR:"):
|
||||
print(result)
|
||||
return 1
|
||||
|
||||
result_dict = json.loads(result)
|
||||
@click.command()
|
||||
@click.option(
|
||||
"--mxpass",
|
||||
"mxpassfile",
|
||||
metavar="filepath",
|
||||
default=".mxpass",
|
||||
help="mxpass file name",
|
||||
)
|
||||
@click.argument("mxid", metavar="MatrixID")
|
||||
@catch_exception(handle=(PygomxAPIError))
|
||||
def smalsetup(mxid, mxpassfile):
|
||||
"""Utility for creating smalbot mxpass files"""
|
||||
|
||||
create_mxpass = len(mxpassfile.strip()) > 0
|
||||
|
||||
if create_mxpass:
|
||||
if os.path.exists(mxpassfile):
|
||||
raise click.ClickException(f"file {mxpassfile} exists.")
|
||||
|
||||
result_dict = ApiV0.Discover(mxid)
|
||||
|
||||
result_dict["password"] = getpass.getpass(prompt="Password: ")
|
||||
data = json.dumps(result_dict).encode(encoding="utf-8")
|
||||
result_dict["make_master_key"] = True
|
||||
result_dict["make_recovery_key"] = True
|
||||
|
||||
r = lib.apiv0_login(data)
|
||||
result = ffi.string(r).decode("utf-8")
|
||||
lib.FreeCString(r)
|
||||
now = int(time.time())
|
||||
result_dict["deviceid"] = f"smalbot-{now}"
|
||||
result_dict["devicename"] = f"smalbot-{datetime.fromtimestamp(now)}"
|
||||
|
||||
if result.startswith("ERR:"):
|
||||
print(result)
|
||||
return 1
|
||||
ApiV0.Login(result_dict, ".mxpass")
|
||||
|
||||
# Set restrictive umask (owner only)
|
||||
new_umask = 0o077
|
||||
old_umask = os.umask(new_umask)
|
||||
|
||||
# Create file with new umask
|
||||
with open(".mxpass", "w") as f:
|
||||
f.write(result)
|
||||
|
||||
# Restore original umask
|
||||
os.umask(old_umask)
|
||||
|
||||
print("login created. start your bot now.")
|
||||
return 0
|
||||
click.echo("login created. start your bot now.")
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue