fix: Invalidate Expired Shared Links (#5065)

mealie/app.py

@@ -124,6 +124,7 @@ register_debug_handler(app)
 
 async def start_scheduler():
     SchedulerRegistry.register_daily(
+        tasks.purge_expired_tokens,
         tasks.purge_group_registration,
         tasks.purge_password_reset_tokens,
         tasks.purge_group_data_exports,

mealie/routes/recipe/shared_routes.py

@@ -2,6 +2,7 @@ from fastapi import APIRouter, Depends, HTTPException
 from pydantic import UUID4
 from sqlalchemy.orm.session import Session
 
+from mealie.core.root_logger import get_logger
 from mealie.db.db_setup import generate_session
 from mealie.repos.all_repositories import get_repositories
 from mealie.schema.recipe import Recipe
@@ -9,12 +10,22 @@ from mealie.schema.response import ErrorResponse
 
 router = APIRouter()
 
+logger = get_logger()
+
 
 @router.get("/shared/{token_id}", response_model=Recipe)
 def get_shared_recipe(token_id: UUID4, session: Session = Depends(generate_session)):
     db = get_repositories(session, group_id=None, household_id=None)
 
     token_summary = db.recipe_share_tokens.get_one(token_id)
+    if token_summary and token_summary.is_expired:
+        try:
+            db.recipe_share_tokens.delete(token_id)
+            session.commit()
+        except Exception:
+            logger.exception(f"Failed to delete expired token {token_id}")
+            session.rollback()
+        token_summary = None
 
     if token_summary is None:
         raise HTTPException(status_code=404, detail=ErrorResponse.respond("Token Not Found"))

mealie/routes/shared/__init__.py

@@ -3,6 +3,7 @@ from functools import cached_property
 from fastapi import HTTPException
 from pydantic import UUID4
 
+from mealie.repos.all_repositories import get_repositories
 from mealie.routes._base import BaseUserController, controller
 from mealie.routes._base.mixins import HttpRepo
 from mealie.routes._base.routers import UserAPIRouter
@@ -32,7 +33,8 @@ class RecipeSharedController(BaseUserController):
     @router.post("", response_model=RecipeShareToken, status_code=201)
     def create_one(self, data: RecipeShareTokenCreate) -> RecipeShareToken:
         # check if recipe group id is the same as the user group id
-        recipe = self.repos.recipes.get_one(data.recipe_id, "id")
+        group_repos = get_repositories(self.repos.session, group_id=self.group_id, household_id=None)
+        recipe = group_repos.recipes.get_one(data.recipe_id, "id")
         if recipe is None or recipe.group_id != self.group_id:
             raise HTTPException(status_code=404, detail="Recipe not found in your group")
 

mealie/schema/recipe/recipe_share_token.py

@@ -18,6 +18,10 @@ class RecipeShareTokenCreate(MealieModel):
     recipe_id: UUID4
     expires_at: datetime = Field(default_factory=defaut_expires_at_time)
 
+    @property
+    def is_expired(self) -> bool:
+        return self.expires_at < datetime.now(UTC)
+
 
 class RecipeShareTokenSave(RecipeShareTokenCreate):
     group_id: UUID4

mealie/services/scheduler/tasks/__init__.py

@@ -1,6 +1,7 @@
 from .create_timeline_events import create_mealplan_timeline_events
 from .delete_old_checked_shopping_list_items import delete_old_checked_list_items
 from .post_webhooks import post_group_webhooks
+from .purge_expired_share_tokens import purge_expired_tokens
 from .purge_group_exports import purge_group_data_exports
 from .purge_password_reset import purge_password_reset_tokens
 from .purge_registration import purge_group_registration
@@ -10,6 +11,7 @@ __all__ = [
     "create_mealplan_timeline_events",
     "delete_old_checked_list_items",
     "post_group_webhooks",
+    "purge_expired_tokens",
     "purge_password_reset_tokens",
     "purge_group_data_exports",
     "purge_group_registration",

mealie/services/scheduler/tasks/purge_expired_share_tokens.py

@@ -0,0 +1,19 @@
+from datetime import UTC, datetime
+
+from mealie.db.db_setup import session_context
+from mealie.repos.all_repositories import get_repositories
+from mealie.schema.response.pagination import PaginationQuery
+
+
+def purge_expired_tokens() -> None:
+    current_time = datetime.now(UTC)
+
+    with session_context() as session:
+        db = get_repositories(session, group_id=None)
+        tokens_response = db.recipe_share_tokens.page_all(
+            PaginationQuery(page=1, per_page=-1, query_filter=f"expiresAt < {current_time}")
+        )
+        if not (tokens := tokens_response.items):
+            return
+
+        db.recipe_share_tokens.delete_many([token.id for token in tokens])

tests/integration_tests/user_recipe_tests/test_recipe_share_tokens.py

@@ -1,4 +1,5 @@
 from collections.abc import Generator
+from datetime import UTC, datetime, timedelta
 
 import pytest
 import sqlalchemy
@@ -119,3 +120,52 @@ def test_share_recipe_from_different_group(api_client: TestClient, unique_user:
 
     response = api_client.post(api_routes.shared_recipes, json={"recipeId": str(recipe.id)}, headers=g2_user.token)
     assert response.status_code == 404
+
+
+def test_share_recipe_from_different_household(
+    api_client: TestClient, unique_user: TestUser, h2_user: TestUser, slug: str
+):
+    database = unique_user.repos
+    recipe = database.recipes.get_one(slug)
+    assert recipe
+
+    response = api_client.post(api_routes.shared_recipes, json={"recipeId": str(recipe.id)}, headers=h2_user.token)
+    assert response.status_code == 201
+
+
+def test_get_recipe_from_token(api_client: TestClient, unique_user: TestUser, slug: str):
+    database = unique_user.repos
+    recipe = database.recipes.get_one(slug)
+    assert recipe
+
+    token = database.recipe_share_tokens.create(
+        RecipeShareTokenSave(recipe_id=recipe.id, group_id=unique_user.group_id)
+    )
+
+    response = api_client.get(api_routes.recipes_shared_token_id(token.id))
+    assert response.status_code == 200
+
+    response_data = response.json()
+    assert response_data["id"] == str(recipe.id)
+
+
+def test_get_recipe_from_expired_token_deletes_token_and_returns_404(
+    api_client: TestClient, unique_user: TestUser, slug: str
+):
+    database = unique_user.repos
+    recipe = database.recipes.get_one(slug)
+    assert recipe
+
+    token = database.recipe_share_tokens.create(
+        RecipeShareTokenSave(
+            recipe_id=recipe.id, group_id=unique_user.group_id, expiresAt=datetime.now(UTC) - timedelta(minutes=1)
+        )
+    )
+    fetch_token = database.recipe_share_tokens.get_one(token.id)
+    assert fetch_token
+
+    response = api_client.get(api_routes.recipes_shared_token_id(token.id), headers=unique_user.token)
+    assert response.status_code == 404
+
+    fetch_token = database.recipe_share_tokens.get_one(token.id)
+    assert fetch_token is None

tests/unit_tests/services_tests/scheduler/tasks/test_purge_expired_share_tokens.py

@@ -0,0 +1,38 @@
+from datetime import UTC, datetime, timedelta
+
+from mealie.schema.recipe.recipe import Recipe
+from mealie.schema.recipe.recipe_share_token import RecipeShareTokenSave
+from mealie.services.scheduler.tasks.purge_expired_share_tokens import purge_expired_tokens
+from tests.utils.factories import random_string
+from tests.utils.fixture_schemas import TestUser
+
+
+def test_no_expired_tokens():
+    # make sure this task runs successfully even if there are no expired tokens
+    purge_expired_tokens()
+
+
+def test_delete_expired_tokens(unique_user: TestUser):
+    db = unique_user.repos
+    recipe = db.recipes.create(
+        Recipe(user_id=unique_user.user_id, group_id=unique_user.group_id, name=random_string(20))
+    )
+    assert recipe and recipe.id
+    good_token = db.recipe_share_tokens.create(
+        RecipeShareTokenSave(
+            recipe_id=recipe.id, group_id=unique_user.group_id, expires_at=datetime.now(UTC) + timedelta(hours=1)
+        )
+    )
+    bad_token = db.recipe_share_tokens.create(
+        RecipeShareTokenSave(
+            recipe_id=recipe.id, group_id=unique_user.group_id, expires_at=datetime.now(UTC) - timedelta(hours=1)
+        )
+    )
+
+    assert db.recipe_share_tokens.get_one(good_token.id)
+    assert db.recipe_share_tokens.get_one(bad_token.id)
+
+    purge_expired_tokens()
+
+    assert db.recipe_share_tokens.get_one(good_token.id)
+    assert not db.recipe_share_tokens.get_one(bad_token.id)