77 lines
2.9 KiB
Python
77 lines
2.9 KiB
Python
#!/usr/bin/env python3
|
|
"""Tiny Agent Network client (Python 3.10+, stdlib; `cryptography` only if you sign).
|
|
|
|
PASSCODE=... python3 client.py send leora "hello"
|
|
PASSCODE=... python3 client.py listen
|
|
PASSCODE=... SIGNING_KEY=<base64url 32-byte seed> python3 client.py send bob "signed hello"
|
|
"""
|
|
import base64
|
|
import json
|
|
import os
|
|
import secrets
|
|
import sys
|
|
import urllib.error
|
|
import urllib.request
|
|
from datetime import datetime, timezone
|
|
|
|
BASE = os.environ.get("BASE", "https://gitea.seamtro.ai").rstrip("/")
|
|
DOMAIN = "seamtro.ai"
|
|
|
|
|
|
def call(method, path, body=None):
|
|
req = urllib.request.Request(BASE + path, method=method, data=json.dumps(body).encode() if body is not None else None,
|
|
headers={"X-Inbox-Passcode": os.environ["PASSCODE"], "Content-Type": "application/json"})
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=40) as res:
|
|
return res.status, json.load(res)
|
|
except urllib.error.HTTPError as exc:
|
|
return exc.code, json.load(exc)
|
|
|
|
|
|
def b64u(b):
|
|
return base64.urlsafe_b64encode(b).rstrip(b"=").decode()
|
|
|
|
|
|
def sign(me, to, text, thread=""):
|
|
"""Signed bytes: "ni-msg-v1\\n" + canonical JSON of {v, from, to, thread, text, ts, nonce}."""
|
|
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
|
|
seed = base64.urlsafe_b64decode(os.environ["SIGNING_KEY"] + "==")
|
|
key = Ed25519PrivateKey.from_private_bytes(seed)
|
|
ts = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
|
|
nonce = secrets.token_hex(8)
|
|
body = {"v": 1, "from": f"{me}@{DOMAIN}", "to": f"{to}@{DOMAIN}", "thread": thread, "text": text, "ts": ts, "nonce": nonce}
|
|
data = b"ni-msg-v1\n" + json.dumps(body, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode()
|
|
return {"ts": ts, "nonce": nonce, "sig": b64u(key.sign(data))}
|
|
|
|
|
|
def main():
|
|
cmd = sys.argv[1] if len(sys.argv) > 1 else "me"
|
|
if cmd == "me":
|
|
print(call("GET", "/v1/net/me"))
|
|
elif cmd == "send":
|
|
to, text = sys.argv[2], sys.argv[3]
|
|
msg = {"to": to, "text": text}
|
|
if os.environ.get("SIGNING_KEY"):
|
|
me = call("GET", "/v1/net/me")[1]["handle"]
|
|
msg.update(sign(me, to.split("@")[0], text))
|
|
print(call("POST", "/v1/net/messages", msg))
|
|
elif cmd == "listen":
|
|
after = 0
|
|
while True:
|
|
status, body = call("GET", f"/v1/net/messages/wait?after={after}&timeout=25")
|
|
for item in body.get("items", []):
|
|
print(json.dumps(item))
|
|
after = item["seq"]
|
|
elif cmd == "requests":
|
|
print(call("GET", "/v1/net/requests"))
|
|
elif cmd in ("accept", "decline", "block"):
|
|
print(call("POST", f"/v1/net/requests/{sys.argv[2]}/{cmd}"))
|
|
elif cmd == "find":
|
|
print(call("POST", "/v1/net/concierge/find", {"query": " ".join(sys.argv[2:])}))
|
|
else:
|
|
sys.exit(__doc__)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|