add token checking

foxnouns/db/aio.py

@@ -1,6 +1,15 @@
+from sqlalchemy import URL
 from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker
 
-from foxnouns.settings import ASYNC_DATABASE_URL
+from foxnouns.settings import DATABASE
+
+ASYNC_DATABASE_URL = URL.create(
+    "postgresql+asyncpg",
+    username=DATABASE["USER"],
+    password=DATABASE["PASSWORD"],
+    host=DATABASE["HOST"],
+    database=DATABASE["NAME"],
+)
 
 engine = create_async_engine(ASYNC_DATABASE_URL)
 async_session = async_sessionmaker(engine, expire_on_commit=False)

foxnouns/db/sync.py

@@ -1,5 +1,13 @@
-from sqlalchemy import create_engine
+from sqlalchemy import URL, create_engine
 
-from foxnouns.settings import SYNC_DATABASE_URL
+from foxnouns.settings import DATABASE
+
+SYNC_DATABASE_URL = URL.create(
+    "postgresql+psycopg",
+    username=DATABASE["USER"],
+    password=DATABASE["PASSWORD"],
+    host=DATABASE["HOST"],
+    database=DATABASE["NAME"],
+)
 
 engine = create_engine(SYNC_DATABASE_URL)

foxnouns/db/user.py

@@ -1,7 +1,6 @@
 from datetime import datetime
 import enum
 
-from itsdangerous.url_safe import URLSafeTimedSerializer
 from sqlalchemy import Text, Integer, BigInteger, ForeignKey, DateTime
 from sqlalchemy.dialects.postgresql import ARRAY
 from sqlalchemy.orm import Mapped, mapped_column, relationship
@@ -58,9 +57,6 @@ class Token(Base):
         # This way, both `user` and `user.edit` tokens will grant access to `user.edit` endpoints.
         return scope in self.scopes or scope.split(".")[0] in self.scopes
 
-    def token_str(self):
-        ...
-
 
 class AuthType(enum.IntEnum):
     DISCORD = 1

foxnouns/db/util.py

@@ -1,9 +1,14 @@
+import datetime
+
+from itsdangerous import BadSignature
+from itsdangerous.url_safe import URLSafeTimedSerializer
 from sqlalchemy.ext.asyncio import AsyncSession
-from sqlalchemy import select
+from sqlalchemy import select, insert
 from quart import g
 
-from .user import User
-from foxnouns.exceptions import ForbiddenError
+from .user import User, Token
+from foxnouns.exceptions import ForbiddenError, ErrorCode
+from foxnouns.settings import SECRET_KEY
 
 
 async def user_from_ref(session: AsyncSession, user_ref: str):
@@ -25,3 +30,39 @@ async def user_from_ref(session: AsyncSession, user_ref: str):
             query = query.where(User.username == user_ref)
 
     return await session.scalar(query)
+
+
+serializer = URLSafeTimedSerializer(SECRET_KEY)
+
+
+def generate_token(token: Token):
+    return serializer.dumps(token.id)
+
+
+async def create_token(session: AsyncSession, user: User, scopes: list[str] = ["*"]):
+    expires = datetime.datetime.now() + datetime.timedelta(days=90)
+    query = (
+        insert(Token)
+        .values(user_id=user.id, expires_at=expires, scopes=scopes)
+        .returning(Token)
+    )
+    return await session.scalar(query)
+
+
+async def validate_token(session: AsyncSession, header: str) -> (Token, User):
+    try:
+        token_id = serializer.loads(header)
+    except BadSignature:
+        raise ForbiddenError("Invalid token", type=ErrorCode.InvalidToken)
+
+    row = (await session.execute(
+        select(Token, User).join(Token.user).where(Token.id == token_id)
+    )).first()
+
+    if not row or not row.Token:
+        raise ForbiddenError("Invalid token", type=ErrorCode.InvalidToken)
+
+    if row.Token.expires_at < datetime.datetime.now():
+        raise ForbiddenError("Token has expired", type=ErrorCode.InvalidToken)
+
+    return (row.Token, row.User)