Merge pull request 'Added API key requirement to main paths and support for different users' (#1) from feature/user-id-support into main

Reviewed-on: #1

database.sql

@@ -0,0 +1,12 @@
+CREATE TABLE `seadex_state` (
+  `id` int(10) unsigned NOT NULL AUTO_INCREMENT,
+  `user_id` bigint(20) NOT NULL,
+  `anilist_id` int(10) unsigned NOT NULL,
+  `anime_name` varchar(255) NOT NULL,
+  `last_updated` varchar(40) DEFAULT NULL,
+  `active` tinyint(1) NOT NULL DEFAULT '1',
+  `created_at` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP,
+  PRIMARY KEY (`id`),
+  UNIQUE KEY `uq_user_anilist` (`user_id`,`anilist_id`),
+  CONSTRAINT `fk_seadex_state_user` FOREIGN KEY (`user_id`) REFERENCES `users` (`user_id`) ON DELETE CASCADE
+)

requirements.txt

@@ -12,8 +12,9 @@ httpcore==1.0.9
 httpx==0.28.1
 idna==3.20
 lib-db-module==0.3.0
-lib-secret-manager==0.3.1
+lib-secret-manager==0.3.6
 lib-send-notifications==0.1.0
+lib-service-auth==0.1.0
 msgspec==0.22.0
 mysql-connector-python==9.4.0
 opentelemetry-api==1.45.1

src/auth_config.py

@@ -0,0 +1,13 @@
+from enum import StrEnum
+
+from service_auth import KeySource, ServiceAuth
+
+
+class Caller(StrEnum):
+	GATEWAY = "gateway"
+	SCHEDULER = "scheduler"
+
+auth = ServiceAuth({
+	Caller.GATEWAY: KeySource("GATEWAY_KEY", "/etc/secrets/gateway_key"),
+	Caller.SCHEDULER: KeySource("SCHEDULER_KEY", "/etc/secrets/scheduler_key"),
+})

src/main.py

@@ -4,11 +4,13 @@ from contextlib import asynccontextmanager
 from datetime import UTC, datetime
 
 import uvicorn
-from fastapi import Depends, FastAPI, HTTPException, Request
+from fastapi import APIRouter, Depends, FastAPI, HTTPException, Request
 from pydantic import BaseModel, Field, computed_field
 from seadex import EntryNotFoundError, SeaDexEntry
+from service_auth import UserId
 
 from anilist_query import AniListError, AnimeNotFoundError, MissingAnimeName, get_anime_name
+from auth_config import Caller, auth
 from db_module import (
 	close_connection_pool,
 	create_connection_pool,
@@ -70,9 +72,13 @@ api = FastAPI(
 		title="Service SeaDex",
 		description="Queries SeaDex for new torrent information",
 		version="1.0.0",
-		lifespan=lifespan
+		lifespan=lifespan,
+		dependencies=[Depends(auth.authenticate)],
 )
 
+user_router = APIRouter(dependencies=[Depends(auth.require(Caller.GATEWAY))])
+scheduler_router = APIRouter(dependencies=[Depends(auth.require(Caller.SCHEDULER))])
+
 @api.middleware("http")
 async def prometheus_middleware(request: Request, call_next):
 	logger.debug(f"[Metrics] Incoming request: {request.method} {request.url.path}")
@@ -101,22 +107,25 @@ def check_for_update(seadex: SeaDexEntry, anilist_id: int) -> str | None:
 		return None
 
 
-@api.post("/update")
+@scheduler_router.post("/update")
 def check_seadex(request:Request, db=Depends(get_db)):
 	results = {"checked": 0, "new": 0, "errors": 0}
 
 	with db.cursor() as cursor:
 		cursor.execute(
-			"SELECT id, anilist_id, anime_name, last_updated "
+			"SELECT id, anilist_id, anime_name, last_updated, user_id "
 			"FROM seadex_state WHERE active = 1"
 		)
 		anime = cursor.fetchall()
 
+	cache: dict[int, str | None] = {}
 	with SeaDexEntry() as seadex:
-		for row_id, anilist_id, anime_name, last_updated in anime:
+		for row_id, anilist_id, anime_name, last_updated, user_id in anime:
 			results["checked"] += 1
 			try:
-				current = check_for_update(seadex, anilist_id)
+				if anilist_id not in cache:
+					cache[anilist_id] = check_for_update(seadex, anilist_id)
+				current = cache[anilist_id]
 
 				if current is None:
 					logger.debug(f"[FastAPI] No entry yet for {anime_name} on SeaDex")
@@ -124,31 +133,32 @@ def check_seadex(request:Request, db=Depends(get_db)):
 
 				if current != last_updated:
 					logger.info(
-						f"[FastAPI] {anime_name} updated, Anilist ID: {anilist_id}"
+						f"[FastAPI] {anime_name} updated, Anilist ID: {anilist_id}, User ID: {user_id}"
 					)
 					request.app.state.notifier.send(
 						title=anime_name,
 						body="SeaDex entry updated",
 						link=f"https://seadex.moe/{anilist_id}/",
+						user_id=user_id,
 					)
 
 					with db.cursor() as cursor:
 						cursor.execute(
-							"UPDATE seadex_state SET last_updated = %s WHERE id = %s",
+							"UPDATE seadex_state SET last_updated = %s WHERE id = %s AND user_id = %s",
-							(current, row_id),
+							(current, row_id, user_id),
 						)
 					db.commit()
 
 					logger.debug(
 						f"[FastAPI] Notified and updated db for id={row_id}, "
-						f"Anilist ID={anilist_id}"
+						f"Anilist ID={anilist_id}, User ID: {user_id}"
 					)
 					results["new"] += 1
 
 			except Exception:
 				db.rollback()
 				results["errors"] += 1
-				logger.exception(f"[FastAPI] Failed for id={row_id} Anilist={anilist_id}",)
+				logger.exception(f"[FastAPI] Failed for id={row_id} Anilist={anilist_id}, User ID: {user_id}",)
 
 	if results["errors"]:
 		logger.warning(f"[FastAPI] SeaDex check finished with errors: {results}")
@@ -157,9 +167,9 @@ def check_seadex(request:Request, db=Depends(get_db)):
 	logger.info(f"[FastAPI] SeaDex check complete: {results}")
 	return results
 
-@api.post("/add/anime")
+@user_router.post("/add/anime")
-def add_anime(anime: AnimeCreate, db=Depends(get_db)):
+def add_anime(anime: AnimeCreate, user_id: UserId, db=Depends(get_db)):
-	logger.info(f"[FastAPI] Received request to track anilist_id={anime.anilist_id}")
+	logger.info(f"[FastAPI] Received request to track anilist_id={anime.anilist_id}, user_id={user_id}")
  
 	try:
 		anime_name = get_anime_name(anime.anilist_id)
@@ -175,16 +185,16 @@ def add_anime(anime: AnimeCreate, db=Depends(get_db)):
 	with db.cursor() as cursor:
 		try:
 			cursor.execute(
-				"SELECT id FROM seadex_state WHERE anilist_id = %s",
+				"SELECT id FROM seadex_state WHERE anilist_id = %s AND user_id = %s",
-				(anime.anilist_id,),
+				(anime.anilist_id, user_id),
 			)
  
 			if cursor.fetchone():
 				logger.debug("[FastAPI] Found existing entry, updating name and setting to active")
 				cursor.execute(
 					"UPDATE seadex_state SET anime_name = %s, active = 1 "
-					"WHERE anilist_id = %s",
+					"WHERE anilist_id = %s AND user_id=%s",
-					(anime_name, anime.anilist_id),
+					(anime_name, anime.anilist_id, user_id),
 				)
 				db.commit()
 				return {
@@ -195,12 +205,12 @@ def add_anime(anime: AnimeCreate, db=Depends(get_db)):
  
 			with SeaDexEntry() as seadex:
 				current = check_for_update(seadex, anime.anilist_id)
-			logger.debug("[FastAPI] Inserting new anime with last_updated={current}")
+			logger.debug(f"[FastAPI] Inserting new anime with last_updated={current}")
 
 			cursor.execute(
-				"INSERT INTO seadex_state (anilist_id, anime_name, last_updated) "
+				"INSERT INTO seadex_state (anilist_id, anime_name, last_updated, user_id) "
-				"VALUES (%s, %s, %s)",
+				"VALUES (%s, %s, %s, %s)",
-				(anime.anilist_id, anime_name, current),
+				(anime.anilist_id, anime_name, current, user_id),
 			)
 			db.commit()
 			return {
@@ -211,27 +221,27 @@ def add_anime(anime: AnimeCreate, db=Depends(get_db)):
  
 		except Exception as e:
 			db.rollback()
-			logger.exception(f"[FastAPI] Failed for anilist_id={anime.anilist_id} name={anime_name}")
+			logger.exception(f"[FastAPI] Failed for anilist_id={anime.anilist_id}, name={anime_name}, user_id={user_id}")
 			raise HTTPException(status_code=500) from e
 
-@api.post("/delete/anime")
+@user_router.post("/delete/anime")
-def remove_anime(anime: AnimeCreate, db=Depends(get_db)):
+def remove_anime(anime: AnimeCreate, user_id: UserId, db=Depends(get_db)):
-	logger.info(f"[FastAPI] Received request to deactivate tracking for anilist_id={anime.anilist_id}")
+	logger.info(f"[FastAPI] Received request to deactivate tracking for anilist_id={anime.anilist_id}, user_id={user_id}")
 	with db.cursor() as cursor:
 			try:
 				cursor.execute(
-					"SELECT active FROM seadex_state WHERE anilist_id = %s",
+					"SELECT active FROM seadex_state WHERE anilist_id = %s AND user_id = %s",
-					(anime.anilist_id,),
+					(anime.anilist_id, user_id),
 				)
 				row = cursor.fetchone()
 				if row and row[0]:
 					cursor.execute(
 						"UPDATE seadex_state SET active = 0 "
-						"WHERE anilist_id = %s",
+						"WHERE anilist_id = %s AND user_id = %s",
-						(anime.anilist_id,),
+						(anime.anilist_id, user_id),
 					)
 					db.commit()
-					logger.debug(f"[FastAPI] Successfully deactivated tracking for anilist_id={anime.anilist_id}")
+					logger.debug(f"[FastAPI] Successfully deactivated tracking for anilist_id={anime.anilist_id}, user_id={user_id}")
 					return {
 						"status": "ok",
 						"detail": f"Deactivated entry {anime.anilist_id}",
@@ -239,28 +249,30 @@ def remove_anime(anime: AnimeCreate, db=Depends(get_db)):
 
 			except Exception as e:
 				db.rollback()
-				logger.exception(f"[FastAPI] Failed removeing tracking for id={anime.anilist_id}")
+				logger.exception(f"[FastAPI] Failed removeing tracking for id={anime.anilist_id}, user_id={user_id}")
 				raise HTTPException(status_code=500) from e
 
 	if row is None:
-		logger.debug(f"[FastAPI] No tracked entry for anilist_id={anime.anilist_id}")
+		logger.debug(f"[FastAPI] No tracked entry for anilist_id={anime.anilist_id}, user_id={user_id}")
 		raise HTTPException(status_code=404, detail="No tracked entry for this AniList ID")
 	if not row[0]:
-		logger.debug(f"[FastAPI] Tracking was already deactivated for anilist_id={anime.anilist_id}")
+		logger.debug(f"[FastAPI] Tracking was already deactivated for anilist_id={anime.anilist_id}, user_id={user_id}")
 		return {"status": "ok", "detail": f"Entry {anime.anilist_id} was already inactive"}
 
-@api.get("/list/anime", response_model=AnimeList)
+@user_router.get("/list/anime", response_model=AnimeList)
-def list_active_and_inactive_anime(db=Depends(get_db)):
+def list_active_and_inactive_anime(user_id: UserId, db=Depends(get_db)):
-	logger.info(f"[FastAPI] Received request to list anime")
+	logger.info(f"[FastAPI] Received request to list anime for User ID = {user_id}")
 	with db.cursor() as cursor:
 		try:
 			cursor.execute(
 				"SELECT anime_name, active, last_updated " \
-				"FROM seadex_state ORDER BY anime_name"
+				"FROM seadex_state WHERE user_id = %s "
+				"ORDER BY anime_name",
+				(user_id,),
 			)
 			rows = cursor.fetchall()
 		except Exception as e:
-			logger.exception(f"[FastAPI] Failed to list anime")
+			logger.exception("[FastAPI] Failed to list anime")
 			raise HTTPException(status_code=500) from e
 
 		anime_list=AnimeList ()
@@ -272,11 +284,16 @@ def list_active_and_inactive_anime(db=Depends(get_db)):
 			(anime_list.active if is_active else anime_list.inactive).append(entry)
 		return anime_list
 
-
+api.include_router(user_router)
+api.include_router(scheduler_router)
 
 async def start_servers():
 	logger.info("[Server] Starting main API (port 5000) and metrics server (port 9000)...")
 	logger.info(f"[Server] Build commit id={os.getenv("COMMIT_ID","unknown")}")
+
+	auth.load_keys(api, Caller.GATEWAY, Caller.SCHEDULER)
+	cleanup_secret_files(auth.key_paths(Caller.GATEWAY, Caller.SCHEDULER))
+
 	config_main = uvicorn.Config(api, host="0.0.0.0", port=5000, log_level=LOG_LEVEL.lower(),log_config=None)
 	config_metrics = uvicorn.Config(metrics_api, host="0.0.0.0", port=9000, log_level=LOG_LEVEL.lower(),log_config=None)
 

src/metrics_server.py

@@ -1,10 +1,14 @@
 from fastapi import FastAPI, Response
 from prometheus_client import CONTENT_TYPE_LATEST, Counter, generate_latest
 
-metrics_api = FastAPI(title="Metrics Server", description="Prometheus metrics endpoint")
+metrics_api = FastAPI(
+	title="Metrics Server",
+	description="Prometheus metrics endpoint",
+)
 
 REQUEST_COUNTER = Counter("http_requests_total", "Total HTTP Requests", ["method", "endpoint", "status"])
 
+
 @metrics_api.get("/metrics")
 async def metrics():
 	return Response(generate_latest(), media_type=CONTENT_TYPE_LATEST)