Add pluralkit integration, work on a simple UI for testing

src/db.py

@@ -13,17 +13,22 @@ class User(Model):
         table = "users"
 
     class PydanticMeta:
-        exclude = ("password",)
+        exclude = ("password", "pluralkit_token")
 
     uuid: _uuid.UUID = fields.UUIDField(pk=True)
     """The UUID of this user account. Always unique."""
-    username: str = fields.CharField(max_length=64)
+    username: str = fields.CharField(max_length=100)
     """This account's username. Must be at most 64 characters long."""
     password: str = fields.TextField()
     """The encoded hashed password for this account."""
     admin: bool = fields.BooleanField(default=False)
     """Whether this account has administrative privileges."""
 
+    pluralkit_id: str | None = fields.CharField(max_length=8, null=True)
+    """The linked PluralKit system ID, if any."""
+    pluralkit_token: str | None = fields.CharField(max_length=512, null=True)
+    """The linked PluralKit token, if any."""
+
     members: fields.ReverseRelation["Member"]
 
     def verify_password(self, password: str) -> bool:
@@ -53,7 +58,7 @@ class Member(Model):
         "models.User", related_name="members", on_delete=fields.CASCADE
     )
     """The user account that this member is associated with."""
-    name: str = fields.CharField(max_length=64)
+    name: str = fields.CharField(max_length=100)
     """The name of this member. Must be at most 64 characters long."""
     avatar_hash: str | None = fields.CharField(max_length=512, null=True)
     """The hash of the avatar image for this member, if any."""

src/main.py

@@ -1,5 +1,6 @@
 import contextlib
 import os.path
+from pathlib import Path
 
 from fastapi import FastAPI
 from tortoise import generate_config
@@ -33,6 +34,16 @@ async def init_db(app):
         yield
 
 app = FastAPI(lifespan=init_db, debug=True)
+
+
+@app.get("/mcaptcha-metadata", include_in_schema=False)
+async def mcaptcha_metadata():
+    return {
+        "sitekey": os.getenv("MCAPTCHA_SITEKEY"),
+        "url": os.getenv("MCAPTCHA_URL"),
+    }
+
+
 app.include_router(account_router)
 app.include_router(image_router)
 app.include_router(oauth2_router)
@@ -45,4 +56,5 @@ if os.name == "nt":
     mimetypes.init()
     mimetypes.add_type("image/webp", ".webp")
 
-app.mount("/static", StaticFiles(directory="static", html=True), name="static")
+app.mount("/static", StaticFiles(directory="static"), name="static")
+app.mount("/ui", StaticFiles(directory=Path(__file__).parent / "simple_ui", html=True), name="ui")

src/routes/account.py

@@ -6,9 +6,19 @@ from pathlib import Path
 import PIL.Image
 from typing import Annotated
 
-from fastapi import APIRouter, UploadFile, HTTPException, status
+import httpx
+from fastapi import APIRouter, Form, Query, UploadFile, HTTPException, status
+from tortoise.transactions import in_transaction
+
 from .oauth2 import session_with_scope
-from ..db import OauthToken, User_Pydantic, Member_Pydantic, Member
+from ..db import OauthToken, User, User_Pydantic, Member_Pydantic, Member
+
+
+MCAPTCHA_SITE_KEY = os.getenv("MCAPTCHA_SITE_KEY")
+MCAPTCHA_URL = os.getenv("MCAPTCHA_URL")
+MCAPTCHA_ENABLED = MCAPTCHA_SITE_KEY is not None and MCAPTCHA_URL is not None
+if os.getenv("MCAPTCHA_ENABLED", "").lower() in ("false", "0", "no", "off"):
+    MCAPTCHA_ENABLED = False
 
 
 def convert_to(source: bytes, new_format: str = "webp", quality: int = 50, resize: bool = True) -> bytes:
@@ -40,6 +50,108 @@ async def get_current_account(token: Annotated[OauthToken, session_with_scope("a
     return await User_Pydantic.from_tortoise_orm(token.user)
 
 
+@router.post("/new", status_code=status.HTTP_201_CREATED, dependencies=[session_with_scope("admin")])
+async def create_new_account(
+        username: str = Form(...),
+        password: str = Form(...),
+        admin: bool = Query(False)
+):
+    """Create a new account. This endpoint is administrator only."""
+    user = await User.create(
+        username=username,
+        password=User.encrypt_password(password),
+        admin=admin
+    )
+    return await User_Pydantic.from_tortoise_orm(user)
+
+
+@router.post("/new/from_pluralkit")
+async def create_account_from_pluralkit(
+        pk_token: str = Form(...),
+        mcaptcha__token: str = Form(... if MCAPTCHA_ENABLED else None),
+        import_members: bool = Form(True),
+):
+    """Imports a system from pluralkit. If import_members is false, only the account will be imported."""
+    async with httpx.AsyncClient(
+        headers={
+            "User-Agent": "Mozilla/5.0 (AvatarServe/0.0 +https://git.i-am.nexus/nex/avatar-serve)",
+        }
+    ) as client:
+        if MCAPTCHA_ENABLED:
+            payload = {
+                "token": mcaptcha__token,
+                "key": MCAPTCHA_SITE_KEY
+            }
+            captcha_response = await client.post(MCAPTCHA_URL + "/api/v1/pow/siteverify", json=payload)
+            if captcha_response.status_code != 200:
+                raise HTTPException(
+                    status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+                    detail="Failed to verify captcha"
+                )
+            if not captcha_response.json()["valid"]:
+                raise HTTPException(
+                    status_code=status.HTTP_400_BAD_REQUEST,
+                    detail="Invalid captcha token"
+                )
+        response = await client.get(
+            "https://api.pluralkit.me/v2/systems/@me",
+            headers={"Authorization": pk_token}
+        )
+        if response.status_code == 403:
+            raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid PluralKit token")
+        if response.status_code != 200:
+            raise HTTPException(
+                response.status_code,
+                detail={
+                    "original": response.json(),
+                    "message": "Failed to fetch system information from PluralKit"
+                }
+            )
+        pk_system = response.json()
+        async with in_transaction() as tx:
+            existing = await User.get_or_none(pluralkit_id=pk_system["id"], using_db=tx)
+            if existing:
+                raise HTTPException(
+                    status_code=status.HTTP_409_CONFLICT,
+                    detail="System already imported. Did you want to refresh?"
+                )
+            user = await User.create(
+                username=pk_system["name"],
+                password=User.encrypt_password(pk_token),
+                pluralkit_id=pk_system["id"],
+                pluralkit_token=pk_token,
+                using_db=tx
+            )
+            if import_members:
+                response = await client.get(
+                    f"https://api.pluralkit.me/v2/systems/@me/members",
+                    headers={"Authorization": pk_token}
+                )
+                if response.status_code != 200:
+                    raise HTTPException(
+                        response.status_code,
+                        detail={
+                            "original": response.json(),
+                            "message": "Failed to fetch member information from PluralKit. No changes made."
+                        }
+                    )
+                members = response.json()
+                for member in members:
+                    await Member.create(
+                        user=user,
+                        name=member["name"],
+                        using_db=tx
+                    )
+        return await User_Pydantic.from_tortoise_orm(user)
+
+
+@router.delete("", status_code=status.HTTP_204_NO_CONTENT)
+async def delete_current_account(token: Annotated[OauthToken, session_with_scope("account:delete")]):
+    """Delete the current account"""
+    await token.user.delete()
+    return
+
+
 @router.get("/members")
 async def get_current_account_members(token: Annotated[OauthToken, session_with_scope("account:read")]):
     """Get a list of members in the current account"""

src/routes/oauth2.py

@@ -12,8 +12,9 @@ from ..db import User, OauthToken
 SCOPES = {
     "account:read": "Read account information, including all members",
     "account:write": "Write account information, including adding and removing members",
-    "account:avatar:write": "create/edit/delete avatars for members",
     "account:delete": "Delete the account",
+    "account:avatar:write": "create/edit/delete avatars for members",
+    "account:pluralkit:use": "Use PluralKit integration (refresh, sync to PluralKit)",
     "admin": "Administrator access. Requires an administrator account."
 }
 router = APIRouter(prefix="/api/oauth2", tags=["Oauth2"])

src/simple_ui/import.html

@@ -0,0 +1,17 @@
+<!DOCTYPE html>
+<html lang="en">
+<head>
+    <meta charset="UTF-8">
+    <meta name="viewport" content="width=device-width, initial-scale=1.0">
+    <title>Document</title>
+</head>
+<body>
+    <form>
+        <label>PluralKit token (<code>pk;token</code>)</label>
+        <input type="text" name="pk_token" id="pk_token"><br/>
+        <label>Import members?</label><input type="checkbox" name="import_members" id="import_members" value="true"><br/>
+        <button type="submit" id="submit">Import from pluralkit</button>
+    </form>
+    <script src="import.js"></script>
+</body>
+</html>

src/simple_ui/import.js

@@ -0,0 +1,44 @@
+function importMembers(event) {
+    event.preventDefault();
+    const form = event.target;
+    const pk_token = form[0].value;
+    const import_members = form[1].value;
+
+    const payload = new FormData();
+    payload.append('pk_token', pk_token);
+    payload.append('import_members', import_members);
+
+    form[2].disabled = true;
+    form[2].textContent = "Importing...";
+    fetch(
+        "/api/account/new/from_pluralkit",
+        {
+            method: "POST",
+            body: payload,
+        }
+    ).then(
+        (response) => {
+            if (response.ok) {
+                form[2].textContent = "Imported!";
+                form[2].disabled = false;
+            } else {
+                response.json().then(
+                    (data) => {
+                        console.error(data.detail);
+                        form[2].textContent = "Error: " + JSON.stringify(data.detail);
+                        form[2].disabled = false;
+                    }
+                );
+            }
+        }
+    ).catch(
+        (error) => {
+            console.error(error);
+            form[2].textContent = "Error: " + error;
+            form[2].disabled = false;
+        }
+    );
+}
+
+
+window.addEventListener("load", () => {document.querySelector("form").addEventListener("submit", importMembers);});

src/simple_ui/index.html

@@ -0,0 +1,11 @@
+<!DOCTYPE html>
+<html lang="en">
+<head>
+    <meta charset="UTF-8">
+    <meta name="viewport" content="width=device-width, initial-scale=1.0">
+    <title>Document</title>
+</head>
+<body>
+    
+</body>
+</html>

src/simple_ui/index.js

@@ -0,0 +1,3 @@
+function isLoggedIn() {
+    return true;
+}

src/simple_ui/login.html

@@ -0,0 +1,19 @@
+<!DOCTYPE html>
+<html lang="en">
+<head>
+    <meta charset="UTF-8">
+    <meta name="viewport" content="width=device-width, initial-scale=1.0">
+    <title>Document</title>
+</head>
+<body>
+    <form>
+        <label>Username</label>
+        <input type="text" name="username" id="username"><br/>
+        <label>Password</label>
+        <input type="password" name="password" id="password"><br/>
+        <button type="submit" id="submit">Login</button>
+    </form>
+
+    <script src="login.js"></script>
+</body>
+</html>

src/simple_ui/login.js

@@ -0,0 +1,75 @@
+async function getMyAccount() {
+    const access_token = localStorage.getItem("access_token");
+    if (!access_token) {
+        return;
+    }
+    const response = await fetch("/api/account", {
+        headers: {
+            "Authorization": "Bearer " + access_token,
+        },
+    });
+    if (!response.ok) {
+        return;
+    }
+    return response.json();
+}
+
+
+async function login(username, password, scopes = null) {
+    if(!scopes) {
+        scopes = [
+            "account:read",
+            "account:write",
+            "account:avatar:write",
+        ]
+    }
+    const payload = new FormData();
+    payload.append("username", username);
+    payload.append("password", password);
+    payload.append("scope", scopes.join(" "));
+    payload.append("grant_type", "password");
+
+    const response = await fetch("/api/oauth2/token", {
+        method: "POST",
+        body: payload,
+    });
+    if (!response.ok) {
+        return;
+    }
+    const data = await response.json();
+    localStorage.setItem("access_token", data.access_token);
+    return data;
+}
+
+async function logout() {
+    const access_token = localStorage.getItem("access_token");
+    if (!access_token) {
+        return;
+    }
+    fetch("/api/oauth2/token/revoke", {
+        method: "POST",
+        headers: {
+            "Authorization": "Bearer " + access_token,
+        },
+    });
+    localStorage.removeItem("access_token");
+}
+
+function formCallback(event) {
+    event.preventDefault();
+    const form = event.target;
+    const username = form[0].value;
+    const password = form[1].value;
+    login(username, password).then(
+        (data) => {
+            if (data) {
+                alert("Logged in!");
+                window.location.href = "/";
+            } else {
+                form[2].textContent = "Invalid username or password";
+            }
+        }
+    );
+}
+
+window.addEventListener("load", () => {document.querySelector("form").addEventListener("submit", formCallback);});