Add app tokens

CHANGELOG.md

@@ -2,6 +2,9 @@
 
 ## CURRENT
 
+## v1.2.0
+* Add app token validation
+
 ## v1.1.1
 * Update dependencies
 * Fix issue with empty due dates

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "taskmaster"
-version = "1.1.1"
+version = "1.2.0"
 description = ""
 authors = ["Andres Javier Lopez <code@andresjavierlopez.com>"]
 readme = "README.md"

taskmaster/__main__.py

@@ -20,6 +20,7 @@ def cli():
 
 cli.add_command(auth_cli.add_user)
 cli.add_command(auth_cli.change_user_password)
+cli.add_command(auth_cli.generate_app_token)
 
 
 if __name__ == "__main__":

taskmaster/auth/cli.py

@@ -1,8 +1,10 @@
 import asyncio
+from datetime import timedelta
 from uuid import UUID
 
 import click
 
+from taskmaster.auth.token import Token, create_app_token
 from taskmaster.database.managers import UserManager
 from taskmaster.schemas.users import User
 
@@ -34,3 +36,11 @@ def change_user_password(user_id):
     password = click.prompt("Password", hide_input=True)
     asyncio.run(_change_password(user_id, password))
     click.echo(f"Password for user {user_id} changed")
+
+
+@click.command()
+@click.option("--app-name", prompt="App name")
+def generate_app_token(app_name):
+    token = Token.create_with_username(app_name)
+    app_token = create_app_token(token)
+    click.echo(f"App token for {app_name}: {app_token}")

taskmaster/auth/endpoints.py

@@ -15,6 +15,7 @@
 from .fb import get_authorization_url, get_fb_info, get_fb_info_from_token
 from .token import (
     AccessTokenInput,
+    PhoneTokenInput,
     RefreshTokenInput,
     Token,
     TokenResponse,
@@ -101,6 +102,91 @@ async def refresh_authentication_token(
     )
 
 
+@router.post("/app-token/refresh", response_model=TokenResponse)
+async def get_app_token(refresh_token_input: RefreshTokenInput):
+    credentials_exception = HTTPException(
+        status_code=status.HTTP_401_UNAUTHORIZED,
+        detail="Could not validate credentials",
+        headers={"WWW-Authenticate": "Bearer"},
+    )
+
+    refresh_token = refresh_token_input.refresh_token
+
+    try:
+        token = Token.decode_token(refresh_token)
+    except InvalidTokenError:
+        raise credentials_exception
+
+    if "app-token" not in token.scopes:
+        raise credentials_exception
+
+    if "refresh-token" not in token.scopes:
+        raise credentials_exception
+
+    token_data = Token.create_with_username(token.username)
+    token_data.scopes = ["app-token"]
+    access_token = create_access_token(token_data, fresh=False)
+
+    return TokenResponse(
+        access_token=access_token,
+        refresh_token=None,
+        token_type="bearer",
+    )
+
+
+@router.post("/app-token/user_phone")
+async def get_user_token_with_app_token_and_phone_number(
+    session: DBSession,
+    phone_token_input: PhoneTokenInput,
+):
+    credentials_exception = HTTPException(
+        status_code=status.HTTP_401_UNAUTHORIZED,
+        detail="Could not validate credentials",
+        headers={"WWW-Authenticate": "Bearer"},
+    )
+
+    try:
+        app_token = Token.decode_token(phone_token_input.app_token)
+    except InvalidTokenError:
+        raise credentials_exception
+
+    if "app-token" not in app_token.scopes:
+        raise credentials_exception
+
+    if "refresh-token" in app_token.scopes:
+        # Don't allow refresh token to be used
+        raise credentials_exception
+
+    stmt = select(UserModel).where(
+        UserModel.phone_number == phone_token_input.phone_number
+    )
+    results = await session.execute(stmt)
+    user = results.scalar_one_or_none()
+
+    if not user:
+        # If user does not exists, let's create one with this phone number
+        user = UserModel(
+            name=phone_token_input.username,
+            phone_number=phone_token_input.phone_number,
+        )
+        session.add(user)
+        await session.commit()
+        await session.refresh(user)
+
+    if user.disabled:
+        raise credentials_exception
+
+    token = Token.create_with_username(user.uuid)
+    token.scopes = [app_token.username]
+    access_token = create_access_token(token, fresh=False)
+
+    return TokenResponse(
+        access_token=access_token,
+        refresh_token=None,
+        token_type="bearer",
+    )
+
+
 @router.get("/auth/fb")
 async def facebook_login():
     authorization_url = await get_authorization_url()

taskmaster/auth/token.py

@@ -43,6 +43,12 @@ class RefreshTokenInput(BaseModel):
     refresh_token: str = Field(alias="refreshToken")
 
 
+class PhoneTokenInput(BaseModel):
+    app_token: str = Field(alias="appToken")
+    phone_number: str = Field(alias="phoneNumber")
+    username: str
+
+
 class TokenResponse(BaseModel):
     access_token: str
     refresh_token: str | None
@@ -79,3 +85,14 @@ def create_refresh_token(data: Token, expires_delta: timedelta | None = None):
     to_encode.update({"exp": expire, "scope": scope})
     encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
     return encoded_jwt
+
+
+def create_app_token(data: Token):
+    """Create an app token that doesn't expire"""
+    to_encode = data.model_dump(exclude={"scopes"})
+    scopes = data.scopes.copy()
+    scopes.extend(["refresh-token", "app-token"])
+    scope = " ".join(scopes)
+    to_encode.update({"scope": scope})
+    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
+    return encoded_jwt