Check access before generating download tokens

Add a table, UserOwnedApp, to track which users have bought/otherwise
have access to which apps. Currently there is no way to add entries to
this table; that is part of the payment processing work.

Author: jameswestman

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/main.py

@@ -5,10 +5,14 @@
 
 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,
@@ -204,10 +208,27 @@ def get_summary(appid: str, response: Response):
 
 
 @app.post("/generate-download-token", status_code=200)
-def get_download_token(appids: List[str]):
+def get_download_token(appids: List[str], login=Depends(logins.login_state)):
     """Generates a download token for the given app IDs."""
 
-    # TODO: Check the user has rights to download 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(
         {

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)