from flask import Flask, Response, redirect
from flask import request
from markupsafe import escape
from mastodon import Mastodon
from email.headerregistry import Address
import jwt
from gevent.pywsgi import WSGIServer
import os
import sqlite3
import zulip
import random
import string
SECRET = os.environ["SECRET"]
ZULIP = f"https://{os.environ['ZULIP']}/accounts/login/jwt/"
REDIRECT = f"https://{os.environ['ZULIP']}/fedi-auth/callback"
DB = os.environ.get("DB", "/var/lib/fedi-zulip/db/applications")
scopes = ["read"]
print(f"""
Zulip is: {os.environ['ZULIP']}
DB location is: {DB}
""")
con = sqlite3.connect(DB)
cur = con.cursor()
zulip_client = zulip.Client()
def get_zulip_user(handle):
print(f"Querying Zulip for handle: {handle}")
zulip_client.call_endpoint(
url=f"/users/{handle}",
method="GET"
)
def create_zulip_user(handle):
print(f"Creating Zulip user with handle: {handle}")
password = ''.join(random.choices(string.ascii_uppercase + string.ascii_lowercase + string.digits, k=40))
full_name = handle.split('@')[0]
payload = {
"email": handle,
"password": password,
"full_name": full_name
}
response = zulip_client.create_user(payload)
if response["result"] == "success":
print(f"Zulip user created for handle: {handle}")
return True
elif response["result"] == "error" and response["msg"] == f"Email '{handle}' already in use":
print(response["msg"] + ", this is okay.")
return True
else:
print(response["msg"])
return False
cur.execute("CREATE TABLE IF NOT EXISTS applications(instance TEXT PRIMARY KEY, client TEXT, secret TEXT, disabled BOOLEAN DEFAULT FALSE)")
def get_application(instance):
res = cur.execute("SELECT client, secret FROM applications WHERE instance = ?", [instance])
return res.fetchone();
def set_application(instance, client, secret):
res = cur.execute("INSERT INTO applications(instance, client, secret) values (?, ?, ?)", (instance, client, secret));
con.commit();
app = Flask(__name__)
@app.get("/fedi-auth/")
def index():
return f"""
Login to Pleroma Chat Using Handle
You can use this page to login to {os.environ['ZULIP']} using your
Pleroma, Akkoma or Mastodon handle. Format is nickname@server.tld.
"""
@app.post("/fedi-auth/login")
def login():
print("Login POST", flush=True)
instance = Address(addr_spec=request.form["nickname"]).domain
print(f"Instance is: {instance}.", flush=True)
try:
app = get_application(instance)
if app == None:
print(f"There is no OAuth application for {instance} so creating one.")
(client, secret) = Mastodon.create_app(
"zulip",
scopes=scopes,
redirect_uris=REDIRECT,
api_base_url=f"https://{instance}",
)
app = (client, secret)
set_application(instance, client, secret)
(client, secret) = app
masto = Mastodon(
client_id=client,
client_secret=secret,
api_base_url=f"https://{instance}",
)
print(f"Getting login URI for {instance}.", flush=True)
oauth = Mastodon.auth_request_url(
masto,
scopes=scopes,
force_login=True,
redirect_uris=REDIRECT,
state=instance,
)
print(f"Sending user to: {REDIRECT}", flush=True)
return redirect(oauth)
except Exception as e:
print(e)
return Response("fail", status=400)
@app.get("/fedi-auth/callback")
def callback():
oauth = request.args.get("code")
instance = request.args.get("state")
print(f"oauth: {oauth is not None} instance: {instance is not None}")
if oauth != None and instance != None:
try:
app = get_application(instance)
if app != None:
masto = Mastodon(
client_id=app[0],
client_secret=app[1],
api_base_url=f"https://{instance}",
)
Mastodon.log_in(masto, code=oauth, scopes=scopes)
creds = Mastodon.account_verify_credentials(masto)
if hasattr(creds, "error"):
print(f"Verifying credentials for instance: {instance} error: {creds.error}")
return Response("fail", status=400)
handle = f"{creds.acct}@{instance}"
success = create_zulip_user(handle)
if not success:
return Response("fail", status=400)
token = jwt.encode(
{"email": handle},
SECRET,
algorithm="HS256",
)
return f"""
Please wait while you are logged in...
"""
except Exception as e:
print(e)
pass
print("Some field wasn't set.")
return Response("fail", status=400)