Add endpoint for generating JWTs

alembic/versions/e0e33ce55700_userownedapp.py

@@ -0,0 +1,34 @@
+"""UserOwnedApp
+
+Revision ID: e0e33ce55700
+Revises: 253ab628caf8
+Create Date: 2022-03-17 19:05:06.230760
+
+"""
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision = 'e0e33ce55700'
+down_revision = '253ab628caf8'
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_table('userownedapp',
+    sa.Column('app_id', sa.String(), nullable=False),
+    sa.Column('account', sa.Integer(), nullable=False),
+    sa.Column('created', sa.DateTime(), nullable=False),
+    sa.ForeignKeyConstraint(['account'], ['flathubuser.id'], ondelete='CASCADE'),
+    sa.PrimaryKeyConstraint('app_id', 'account')
+    )
+    # ### end Alembic commands ###
+
+
+def downgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_table('userownedapp')
+    # ### end Alembic commands ###

app/config.py

@@ -36,6 +36,7 @@ class Settings(BaseSettings):
     google_client_secret: str = "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"
     google_return_url: str = "http://localhost:3000/login/google"
     cors_origins: str = "http://localhost:3000"
+    flat_manager_secret: str = "c2VjcmV0"
 
 
 settings = Settings()

app/main.py

@@ -1,10 +1,18 @@
+import base64
+from datetime import datetime, timedelta
 from functools import lru_cache
+from typing import List
 
+import jwt
 import sentry_sdk
-from fastapi import FastAPI, Response
+from fastapi import Depends, FastAPI, Response
 from fastapi.middleware.cors import CORSMiddleware
+from fastapi.responses import JSONResponse
+from fastapi_sqlalchemy import db as sqldb
 from sentry_sdk.integrations.asgi import SentryAsgiMiddleware
 
+from app import models
+
 from . import (
     apps,
     config,
@@ -199,6 +207,44 @@ def get_summary(appid: str, response: Response):
     return None
 
 
+@app.post("/generate-download-token", status_code=200)
+def get_download_token(appids: List[str], login=Depends(logins.login_state)):
+    """Generates a download token for the given app IDs."""
+
+    if not login["state"].logged_in():
+        return JSONResponse({ "detail": "not_logged_in" }, status_code=401)
+    user = login["user"]
+
+    unowned = [
+        app_id
+        for app_id in appids
+        if not models.UserOwnedApp.user_owns_app(sqldb, user, app_id)
+    ]
+
+    if len(unowned) != 0:
+        return JSONResponse(
+            {
+                "detail": "purchase_necessary",
+                "missing_appids": unowned,
+            },
+            status_code=403,
+        )
+
+    encoded = jwt.encode(
+        {
+            "sub": "download",
+            "exp": datetime.utcnow() + timedelta(hours=24),
+            "prefixes": appids,
+        },
+        base64.b64decode(config.settings.flat_manager_secret),
+        algorithm="HS256",
+    )
+
+    return {
+        "token": encoded,
+    }
+
+
 def sort_ids_by_downloads(ids):
     if len(ids) <= 1:
         return ids

app/models.py

@@ -327,3 +327,55 @@ def delete_user(db, user: FlathubUser):
 
 
 FlathubUser.TABLES_FOR_DELETE.append(UserVerifiedApp)
+
+
+class UserOwnedApp(Base):
+    __tablename__ = "userownedapp"
+
+    app_id = Column(String, nullable=False, primary_key=True)
+    account = Column(
+        Integer,
+        ForeignKey(FlathubUser.id, ondelete="CASCADE"),
+        nullable=False,
+        primary_key=True,
+    )
+    created = Column(DateTime, nullable=False)
+
+    @staticmethod
+    def user_owns_app(db, user: FlathubUser, app_id: str):
+        # If the user is trying to download "org.gnome.Clocks.Locale", permission for "org.gnome.Clocks" should suffice.
+        # Note that the authoritative check for this rule is in flat-manager, not here.
+        segments = app_id.split(".")
+        prefixes = [".".join(segments[:i]) for i in range(len(segments) + 1)]
+
+        return (
+            db.session.query(UserOwnedApp)
+            .filter_by(account=user.id)
+            .filter(UserOwnedApp.app_id.in_(prefixes))
+            .first()
+            is not None
+        )
+
+    @staticmethod
+    def all_by_user(db, user: FlathubUser):
+        return db.session.query(UserOwnedApp).filter_by(account=user.id)
+
+    @staticmethod
+    def delete_hash(hasher: utils.Hasher, db, user: FlathubUser):
+        """
+        Add a user's owned apps to the hasher for token generation
+        """
+        apps = [app.app_id for app in UserOwnedApp.all_by_user(db, user)]
+        apps.sort()
+        for app in apps:
+            hasher.add_string(app)
+
+    @staticmethod
+    def delete_user(db, user: FlathubUser):
+        """
+        Delete any app ownerships associated with this user
+        """
+        db.session.execute(delete(UserOwnedApp).where(UserOwnedApp.account == user.id))
+
+
+FlathubUser.TABLES_FOR_DELETE.append(UserOwnedApp)

pyproject.toml

@@ -23,6 +23,7 @@ itsdangerous = "^2.1"
 PyGithub = "^1.55"
 vcrpy = "^4.1.1"
 python-gitlab = "^3.1"
+PyJWT = "^2.3.0"
 
 [tool.poetry.dev-dependencies]
 black = "^22.1"