add ed25519 signing (not used for anything yet)

database.py

@@ -1,10 +1,14 @@
 import pymongo
+import pymongo.errors
 import redis
 import os
 import secrets
+import time
 from radix import Radix
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey, Ed25519PublicKey
 
-from utils import log
+from utils import log, create_ed25519_keys, import_priv_ed25519_key, import_pub_ed25519_key
+from signing import SigningKeys
 
 CURRENT_DB_VERSION = 9
 
@@ -198,20 +202,41 @@
             "last_seen": None,
             "delete_after": None
         })
-    except: pass
+    except pymongo.errors.DuplicateKeyError: pass
 try:
     db.config.insert_one({
         "_id": "migration",
         "database": 1
     })
-except: pass
+except pymongo.errors.DuplicateKeyError: pass
 try:
     db.config.insert_one({
         "_id": "status",
         "repair_mode": False,
         "registration": True
     })
-except: pass
+except pymongo.errors.DuplicateKeyError: pass
+
+
+# Load existing signing keys or create new ones
+# The active private key should be rotated every 10 days by the background thread
+# and public keys older than 90 days should be invalidated.
+_priv_key = Ed25519PrivateKey.generate()
+if db.config.count_documents({"_id": "signing_key"}, limit=1):
+    _priv_key = Ed25519PrivateKey.from_private_bytes(db.config.find_one({"_id": "signing_key"})["raw"])
+else:
+    db.config.update_one({"_id": "signing_key"}, {"$set": {
+        "raw": _priv_key.private_bytes_raw(),
+        "rotated_at": int(time.time())
+    }}, upsert=True)
+    db.pub_signing_keys.insert_one({
+        "raw": _priv_key.public_key().public_bytes_raw(),
+        "created_at": int(time.time())
+    })
+signing_keys = SigningKeys(_priv_key, [
+    Ed25519PublicKey.from_public_bytes(pub_signing_key["raw"])
+    for pub_signing_key in db.pub_signing_keys.find({})
+])
 
 
 # Load netblocks

security.py

@@ -1,11 +1,11 @@
 from hashlib import sha256
-from typing import Optional
+from typing import Optional, Any
 from email.mime.multipart import MIMEMultipart
 from email.mime.text import MIMEText
 from email.utils import formataddr
 import time, requests, os, uuid, secrets, bcrypt, msgpack, jinja2, smtplib
 
-from database import db, rdb
+from database import db, rdb, signing_keys
 from utils import log
 from uploads import clear_files
 
@@ -15,6 +15,7 @@
 """
 
 SENSITIVE_ACCOUNT_FIELDS = {
+    "email",
     "pswd",
     "mfa_recovery_code",
     "tokens",
@@ -261,6 +262,14 @@ def create_user_token(username: str, ip: str, used_token: Optional[str] = None)
     return new_token
 
 
+def create_token(ttype: int, subject: Any, scopes: int, expires_in: Optional[int] = None) -> str:
+    pass
+
+
+def extract_token(token: str) -> tuple[int, Any, int]:
+    pass
+
+
 def update_settings(username, newdata):
     # Check datatype
     if not isinstance(username, str):
@@ -497,6 +506,21 @@ def background_tasks_loop():
 
         log("Running background tasks...")
 
+        # Rotate signing key (every 10 days)
+        if db.config.count_documents({"_id": "signing_key", "rotated_at": {"$lt": int(time.time())-864000}}, limit=1):
+            new_priv_bytes, new_pub_bytes = signing_keys.rotate()
+            db.pub_signing_keys.insert_one({
+                "raw": new_pub_bytes,
+                "created_at": int(time.time())
+            })
+            db.config.update_one({"_id": "signing_key"}, {"$set": {
+                "raw": new_priv_bytes,
+                "rotated_at": int(time.time())
+            }}, upsert=True)
+
+        # Delete public signing keys that are older than 90 days
+        db.pub_signing_keys.delete_many({"created_at": {"$lt": int(time.time())-7776000}})
+
         # Delete accounts scheduled for deletion
         for user in db.usersv0.find({"delete_after": {"$lt": int(time.time())}}, projection={"_id": 1}):
             try:

signing.py

@@ -0,0 +1,35 @@
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey, Ed25519PublicKey
+from typing import List
+
+class SigningKeys:
+    def __init__(self, priv_bytes: bytes, pub_bytes: bytes|List[bytes]):
+        self.priv_key = Ed25519PrivateKey.from_private_bytes(priv_bytes)
+        if isinstance(pub_bytes, bytes):
+            pub_bytes = [pub_bytes]
+        self.pub_keys = [
+            Ed25519PublicKey.from_public_bytes(_pub_bytes)
+            for _pub_bytes in pub_bytes
+        ]
+
+    def sign(self, data: bytes) -> bytes:
+        return self.priv_key.sign(data)
+
+    def validate(self, signature: bytes, data: bytes) -> bool:
+        for pub_key in self.pub_keys:
+            try:
+                pub_key.verify(signature, data)
+            except:
+                continue
+            else:
+                return True
+
+        return False
+
+    def rotate(self) -> tuple[bytes, bytes]:
+        new_priv = Ed25519PrivateKey.generate()
+        new_pub = new_priv.public_key()
+
+        self.priv_key = new_priv
+        self.pub_keys.append(new_pub)
+
+        return new_priv.private_bytes_raw(), new_pub.public_bytes_raw()

utils.py

@@ -1,6 +1,4 @@
-from typing import Literal
 from datetime import datetime
-import time
 import sys
 import traceback