#!/usr/bin/env python3
import json
import urllib.request
import urllib.error
from datetime import datetime, timedelta
from http.server import BaseHTTPRequestHandler, HTTPServer
from urllib.parse import parse_qs
HOMESERVER = "http://192.168.12.151:8008"
PUBLIC_URL = "https://matrix.theocloud.dev"
ADMIN_USER = "admin"
HTML = """
Matrix Einladung
Matrix Einladungslink
{content}
"""
def login(password: str) -> str:
data = json.dumps({
"type": "m.login.password",
"user": ADMIN_USER,
"password": password,
}).encode()
req = urllib.request.Request(
f"{HOMESERVER}/_matrix/client/v3/login",
data=data,
headers={"Content-Type": "application/json"},
)
with urllib.request.urlopen(req) as resp:
return json.loads(resp.read())["access_token"]
def create_token(access_token: str, uses: int, expires_in_hours: int | None) -> str:
payload: dict = {"uses_allowed": uses}
if expires_in_hours:
expiry = datetime.now() + timedelta(hours=expires_in_hours)
payload["expiry_time"] = int(expiry.timestamp() * 1000)
data = json.dumps(payload).encode()
req = urllib.request.Request(
f"{HOMESERVER}/_synapse/admin/v1/registration_tokens/new",
data=data,
headers={
"Content-Type": "application/json",
"Authorization": f"Bearer {access_token}",
},
)
with urllib.request.urlopen(req) as resp:
return json.loads(resp.read())["token"]
class Handler(BaseHTTPRequestHandler):
def log_message(self, format, *args):
pass
def send_html(self, content: str, status: int = 200):
body = HTML.format(content=content).encode()
self.send_response(status)
self.send_header("Content-Type", "text/html; charset=utf-8")
self.send_header("Content-Length", len(body))
self.end_headers()
self.wfile.write(body)
def do_GET(self):
self.send_html("")
def do_POST(self):
length = int(self.headers.get("Content-Length", 0))
body = self.rfile.read(length).decode()
params = parse_qs(body)
password = params.get("password", [""])[0]
uses = int(params.get("uses", ["1"])[0] or 1)
expires_raw = params.get("expires", [""])[0].strip()
expires_in_hours = int(expires_raw) if expires_raw else None
try:
access_token = login(password)
token = create_token(access_token, uses, expires_in_hours)
except urllib.error.HTTPError as e:
error_body = e.read().decode()
try:
msg = json.loads(error_body).get("error", error_body)
except Exception:
msg = error_body
self.send_html(f'{msg}
')
return
link = f"{PUBLIC_URL}/#/register?token={token}"
meta_parts = [f"{uses}x nutzbar"]
if expires_in_hours:
expires_at = datetime.now() + timedelta(hours=expires_in_hours)
meta_parts.append(f"gültig bis {expires_at.strftime('%d.%m.%Y %H:%M')}")
self.send_html(f"""
{link}
{' · '.join(meta_parts)}
""")
if __name__ == "__main__":
server = HTTPServer(("0.0.0.0", 8090), Handler)
print("Invite app running on http://0.0.0.0:8090")
server.serve_forever()