-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathregister_bot.py
More file actions
125 lines (103 loc) · 4.75 KB
/
Copy pathregister_bot.py
File metadata and controls
125 lines (103 loc) · 4.75 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
"""Register a bot (or user) via the JuggleIM Server API and print its token.
The token returned here is what `ImBotClient.connect(token)` expects.
Authentication (JuggleIM Server API): each request carries the headers
``appkey``, ``nonce`` (random 0-9999), ``timestamp`` (epoch ms) and
``signature = lowercase_hex( SHA1(AppSecret + nonce + timestamp) )``.
No server credentials are hard-coded — supply them via environment variables or
CLI flags:
export IMBOT_API_URL="https://api.juggleim.com/apigateway"
export IMBOT_APPKEY="<your-appkey>"
export IMBOT_APPSECRET="<your-appsecret>"
# register a bot
python examples/register_bot.py --bot-id my-bot --nickname "My Bot"
# or register a normal user
python examples/register_bot.py --user --user-id u1 --nickname "User 1"
Uses only the Python standard library.
"""
import argparse
import hashlib
import json
import os
import random
import ssl
import sys
import time
import urllib.request
def _auth_headers(appsecret: str, appkey: str) -> dict:
nonce = str(random.randint(0, 9999))
timestamp = str(int(time.time() * 1000))
signature = hashlib.sha1((appsecret + nonce + timestamp).encode("utf-8")).hexdigest()
return {
"Content-Type": "application/json",
"appkey": appkey,
"nonce": nonce,
"timestamp": timestamp,
"signature": signature,
}
def _post(api_url: str, path: str, appkey: str, appsecret: str, body: dict, insecure: bool):
url = api_url.rstrip("/") + path
data = json.dumps(body).encode("utf-8")
req = urllib.request.Request(
url, data=data, headers=_auth_headers(appsecret, appkey), method="POST"
)
ctx = ssl._create_unverified_context() if insecure else None
with urllib.request.urlopen(req, timeout=15, context=ctx) as resp:
return resp.status, resp.read().decode("utf-8")
def main() -> int:
parser = argparse.ArgumentParser(description="Register a JuggleIM bot/user and print its token")
parser.add_argument("--api-url", default=os.getenv("IMBOT_API_URL", ""),
help="Server API base, e.g. https://api.juggleim.com/apigateway (env IMBOT_API_URL)")
parser.add_argument("--appkey", default=os.getenv("IMBOT_APPKEY", ""),
help="App AppKey (env IMBOT_APPKEY)")
parser.add_argument("--appsecret", default=os.getenv("IMBOT_APPSECRET", ""),
help="App AppSecret (env IMBOT_APPSECRET)")
parser.add_argument("--bot-id", default="", help="bot id to register")
parser.add_argument("--user-id", default="", help="user id to register (with --user)")
parser.add_argument("--nickname", default="", help="display name")
parser.add_argument("--user", action="store_true",
help="register a normal user (/users/register) instead of a bot")
parser.add_argument("--insecure", action="store_true",
help="skip TLS verification (workaround for an expired server certificate)")
args = parser.parse_args()
missing = [n for n, v in (("--api-url", args.api_url), ("--appkey", args.appkey),
("--appsecret", args.appsecret)) if not v]
if missing:
print("error: missing required config: " + ", ".join(missing)
+ " (pass the flag or set the matching IMBOT_* env var)", file=sys.stderr)
return 2
if args.user:
ident = args.user_id
if not ident:
print("error: --user-id is required with --user", file=sys.stderr)
return 2
path = "/users/register"
body = {"user_id": ident, "nickname": args.nickname, "ext_fields": {}}
else:
ident = args.bot_id
if not ident:
print("error: --bot-id is required", file=sys.stderr)
return 2
path = "/bots/register"
body = {"bot_id": ident, "nickname": args.nickname, "ext_fields": {}}
try:
status, text = _post(args.api_url, path, args.appkey, args.appsecret, body, args.insecure)
except Exception as exc: # noqa: BLE001
print(f"request failed: {exc}", file=sys.stderr)
if "CERTIFICATE_VERIFY_FAILED" in str(exc):
print("hint: the server's TLS certificate may be invalid/expired; "
"retry with --insecure if you trust the endpoint.", file=sys.stderr)
return 1
try:
payload = json.loads(text)
except ValueError:
print(f"HTTP {status}, non-JSON response: {text}", file=sys.stderr)
return 1
if payload.get("code") != 0:
print(f"register failed: HTTP {status} {text}", file=sys.stderr)
return 1
data = payload.get("data", {})
print(f"user_id: {data.get('user_id', '')}")
print(f"token: {data.get('token', '')}")
return 0
if __name__ == "__main__":
raise SystemExit(main())