AI code review changes round 1

Author: VerdantFox

.github/copilot-instructions.md

@@ -112,3 +112,8 @@ python -m scripts.alembic upgrade head
 - The `logged_in_basic_user` / `logged_in_admin_user` fixtures call `test_client.cookies.clear()` on teardown. If a test logs in **manually** (e.g., `POST /login` or `POST /auth/token`) without using these fixtures, it **must** call `test_client.cookies.clear()` at the end to avoid leaking cookie state into subsequent tests.
 - `BASIC_COOKIE` and `ADMIN_COOKIE` in `tests/__init__.py` are module-level dicts shared across the session. They are cleared between test modules by `_clear_tokens()` called from the `clean_db` fixture teardown.
 - Flash messages are stored in the **session cookie**. A stale session cookie from a previous test (e.g., one that logged in) can cause flash messages from a subsequent test to be lost or doubled.
+
+## DO these things
+
+- Run `pre-commit run --all-files` frequently to apply all linters and formatters locally. The `pre-commit` config is in `.pre-commit-config.yaml`
+- `ai_workspace/` at the project root is in `.gitignore` and safe for writing debug scripts and output files. Use this instead of writing to `/tmp/` or other shared locations.

app/datastore/database.py

@@ -16,6 +16,7 @@
 # connect_args={"check_same_thread": False} for SQLite
 
 ENGINE: AsyncEngine | None = None
+SESSION_MAKER: async_sessionmaker[AsyncSession] | None = None
 
 
 def get_engine(
@@ -26,7 +27,7 @@ def get_engine(
     new: bool = False,
 ) -> AsyncEngine:
     """Return the database engine, creating a new one if needed."""
-    global ENGINE  # noqa: PLW0603 (global-statement)
+    global ENGINE, SESSION_MAKER  # noqa: PLW0603 (global-statement)
     if not new and ENGINE is not None:
         return ENGINE
 
@@ -39,14 +40,21 @@ def get_engine(
         pool_size=pool_size,
         pool_pre_ping=True,
     )
+    SESSION_MAKER = None  # reset when engine changes
     return ENGINE
 
 
+def get_session_maker() -> async_sessionmaker[AsyncSession]:
+    """Return the async session maker, creating a new one if needed."""
+    global SESSION_MAKER  # noqa: PLW0603 (global-statement)
+    if SESSION_MAKER is None:
+        SESSION_MAKER = async_sessionmaker(get_engine(), expire_on_commit=False)
+    return SESSION_MAKER
+
+
 async def get_db_session() -> AsyncGenerator[AsyncSession]:
     """Start a SessionLocal transaction and yield it."""
-    engine = get_engine()
-    async_session = async_sessionmaker(engine, expire_on_commit=False)
-    async with async_session() as session:
+    async with get_session_maker()() as session:
         yield session
 
 

app/services/blog/blog_handler.py

@@ -56,6 +56,9 @@ class SaveBlogResponse(BaseModel, arbitrary_types_allowed=True):
     field_errors: defaultdict[str, list[str]] = defaultdict(list)
 
 
+SaveBlogResponse.model_rebuild()
+
+
 class CommentInputPreview(BaseModel, arbitrary_types_allowed=True):
     """Input for comment preview."""
 
@@ -89,6 +92,9 @@ class SaveCommentResponse(BaseModel, arbitrary_types_allowed=True):
     field_errors: defaultdict[str, list[str]] = defaultdict(list)
 
 
+SaveCommentResponse.model_rebuild()
+
+
 def _get_bp_statement() -> Select:
     return select(db_models.BlogPost).options(
         selectinload(db_models.BlogPost.tags),
@@ -286,29 +292,21 @@ async def _get_bp_from_slug_history(db: AsyncSession, slug: str) -> db_models.Bl
     """Get a blog post from its slug history."""
     try:
         stmt = (
-            select(db_models.OldBlogPostSlug)
-            .options(
-                selectinload(db_models.OldBlogPostSlug.blog_post)
-                .selectinload(db_models.BlogPost.comments)
-                .selectinload(db_models.BlogPostComment.user),
-                selectinload(db_models.OldBlogPostSlug.blog_post).selectinload(
-                    db_models.BlogPost.tags
-                ),
+            _get_bp_statement()
+            .join(
+                db_models.OldBlogPostSlug,
+                db_models.OldBlogPostSlug.blog_post_id == db_models.BlogPost.id,
             )
             .filter(db_models.OldBlogPostSlug.slug == slug)
         )
         result = await db.execute(stmt)
-        slug_object = result.scalars().one()
+        return result.scalars().one()
     except sqlalchemy.exc.NoResultFound as e:
         raise errors.BlogPostNotFoundError from e
-    else:
-        return slug_object.blog_post
 
 
 async def save_blog_post(db: AsyncSession, data: SaveBlogInput) -> SaveBlogResponse:
     """Save a blog post."""
-    SaveBlogResponse.model_rebuild()
-
     field_errors: defaultdict[str, list[str]] = defaultdict(list)
     try:
         blog_post = await _save_bp_to_db(data=data, db=db)
@@ -389,7 +387,7 @@ async def update_existing_bp_fields(  # noqa: C901 (complexity)
         blog_post.series_id = data.series_id
     if blog_post.series_position != data.series_position:
         blog_post.series_position = data.series_position
-    blog_post.updated_timestamp = datetime.now().astimezone(UTC)
+    blog_post.updated_timestamp = datetime.now(UTC)
     return blog_post
 
 
@@ -414,7 +412,7 @@ async def set_new_bp_fields(
     html_description = markdown_parser.markdown_to_html(data.description)
     html_content = markdown_parser.markdown_to_html(data.content)
     tags = await _get_bp_tags(db=db, tags=data.tags)
-    now = datetime.now().astimezone(UTC)
+    now = datetime.now(UTC)
     return db_models.BlogPost(
         title=data.title,
         slug=blog_utils.get_slug(data.title),
@@ -575,7 +573,7 @@ async def commit_media_to_db(  # noqa: PLR0913 (too-many-arguments)
         name=name,
         locations=locations_str,
         media_type=media_type,
-        created_timestamp=datetime.now().astimezone(UTC),
+        created_timestamp=datetime.now(UTC),
         position=position,
     )
     db.add(bp_media_object)
@@ -623,7 +621,7 @@ async def update_existing_comment(
     html_content = generate_comment_html(md_content)
     comment.md_content = md_content
     comment.html_content = html_content
-    comment.updated_timestamp = datetime.now().astimezone(UTC)
+    comment.updated_timestamp = datetime.now(UTC)
     if current_user.is_authenticated:
         comment.user_id = current_user.id
     await db.commit()
@@ -634,7 +632,7 @@ async def update_existing_comment(
 def generate_comment(data: CommentInputPreview) -> db_models.BlogPostComment:
     """Generate a blog post comment."""
     html_content = generate_comment_html(data.content)
-    now = datetime.now().astimezone(UTC)
+    now = datetime.now(UTC)
     return db_models.BlogPostComment(
         blog_post_id=data.bp_id,
         name=data.name,
@@ -680,18 +678,20 @@ def can_edit_comment(
     comment: db_models.BlogPostComment, current_user: db_models.User | UnauthenticatedUser
 ) -> bool:
     """Check if a user can edit this comment."""
-    return comment.user_id == current_user.id or comment.guest_id == current_user.guest_id
+    if current_user.is_authenticated:
+        return comment.user_id == current_user.id
+    return bool(comment.guest_id and comment.guest_id == current_user.guest_id)
 
 
 def can_delete_comment(
     comment: db_models.BlogPostComment, current_user: db_models.User | UnauthenticatedUser
 ) -> bool:
     """Check if a user can delete this comment. Allows admin to delete any comment."""
-    return (
-        comment.user_id == current_user.id
-        or comment.guest_id == current_user.guest_id
-        or current_user.is_admin
-    )
+    if current_user.is_admin:
+        return True
+    if current_user.is_authenticated:
+        return comment.user_id == current_user.id
+    return bool(comment.guest_id and comment.guest_id == current_user.guest_id)
 
 
 async def get_comment_from_id(db: AsyncSession, comment_id: int) -> db_models.BlogPostComment:

app/services/general/encryption_handler.py

@@ -9,8 +9,6 @@
 
 from app.settings import settings
 
-ENCRYPTION_KEY = settings.encryption_key
-
 
 def derive_iv(plaintext: str) -> bytes:
     """Derive a consistent IV from the plaintext using a hash function.
@@ -39,7 +37,7 @@ def encrypt(data: str) -> str:
         str: The Base64 encoded encrypted data including the IV.
 
     """
-    key_bytes = bytes.fromhex(ENCRYPTION_KEY)
+    key_bytes = bytes.fromhex(settings.encryption_key)
     data_bytes = data.encode("utf-8")
     iv = derive_iv(data)
     cipher = Cipher(algorithms.AES(key_bytes), modes.CBC(iv), backend=default_backend())
@@ -68,7 +66,7 @@ def decrypt(encoded_data: str) -> str:
         str: The decrypted data as a string.
 
     """
-    key_bytes = bytes.fromhex(ENCRYPTION_KEY)
+    key_bytes = bytes.fromhex(settings.encryption_key)
     encrypted_data_with_iv = base64.b64decode(encoded_data)
 
     iv = encrypted_data_with_iv[:16]
@@ -89,7 +87,7 @@ def decrypt(encoded_data: str) -> str:
 if __name__ == "__main__":  # pragma: no cover
     # Encryption key as a 64-character hex string (256 bits / 32 bytes)
     # key = os.urandom(32).hex()  # noqa: ERA001 (commented-out code)
-    print(f"Key (hex): {ENCRYPTION_KEY}")  # noqa: T201 (print used for example)
+    print(f"Key (hex): {settings.encryption_key}")  # noqa: T201 (print used for example)
 
     import uuid
 

app/services/general/transforms.py

@@ -18,12 +18,12 @@ def to_bool(value: Any) -> bool:
     return False
 
 
-def too_bool_validator(v: Any, _: ValidationInfo) -> bool:
-    """Cooerce any value to a bool."""
+def to_bool_validator(v: Any, _: ValidationInfo) -> bool:
+    """Coerce any value to a bool."""
     return to_bool(v)
 
 
-CoercedBool = Annotated[bool, BeforeValidator(too_bool_validator)]
+CoercedBool = Annotated[bool, BeforeValidator(to_bool_validator)]
 
 
 def to_list(obj: Any, *, lowercase: bool = False) -> list[str]:

app/services/users/user_handler.py

@@ -170,7 +170,7 @@ async def create_pw_reset_token(
     # Delete any existing tokens older than PW_RESET_TOKEN_EXPIRATION_MINUTES
     stmt = delete(db_models.PasswordResetToken).where(
         db_models.PasswordResetToken.created_timestamp
-        < datetime.now().astimezone(UTC) - timedelta(minutes=PW_RESET_TOKEN_EXPIRATION_MINUTES)
+        < datetime.now(UTC) - timedelta(minutes=PW_RESET_TOKEN_EXPIRATION_MINUTES)
     )
     await db.execute(stmt)
 
@@ -179,9 +179,8 @@ async def create_pw_reset_token(
     pw_reset_token = db_models.PasswordResetToken(
         user_id=user.id,
         encrypted_query=encryption_handler.encrypt(query),
-        created_timestamp=datetime.now().astimezone(UTC),
-        expires_timestamp=datetime.now().astimezone(UTC)
-        + timedelta(minutes=PW_RESET_TOKEN_EXPIRATION_MINUTES),
+        created_timestamp=datetime.now(UTC),
+        expires_timestamp=datetime.now(UTC) + timedelta(minutes=PW_RESET_TOKEN_EXPIRATION_MINUTES),
     )
     db.add(pw_reset_token)
     await db.commit()
@@ -202,26 +201,14 @@ async def assert_token_is_valid(db: AsyncSession, query: str) -> db_models.Passw
     except sqlalchemy.orm.exc.NoResultFound as e:
         raise errors.PasswordResetTokenNotFoundError from e
     token_dt = token.expires_timestamp.astimezone(UTC)
-    if token_dt < datetime.now().astimezone(UTC):
+    if token_dt < datetime.now(UTC):
         raise errors.PasswordResetTokenExpiredError
     return token
 
 
 async def reset_password_from_token(db: AsyncSession, query: str, password: str) -> db_models.User:
     """Reset a user's password from a password reset token."""
-    encrypted_query = encryption_handler.encrypt(query)
-
-    get_token_stmt = select(db_models.PasswordResetToken).where(
-        db_models.PasswordResetToken.encrypted_query == encrypted_query
-    )
-    get_token_result = await db.execute(get_token_stmt)
-    try:
-        pw_reset_token = get_token_result.scalars().one()
-    except sqlalchemy.orm.exc.NoResultFound as e:
-        raise errors.PasswordResetTokenNotFoundError from e
-    token_dt = pw_reset_token.expires_timestamp.astimezone(UTC)
-    if token_dt < datetime.now().astimezone(UTC):
-        raise errors.PasswordResetTokenExpiredError
+    pw_reset_token = await assert_token_is_valid(db, query)
     get_user_stmt = select(db_models.User).where(db_models.User.id == pw_reset_token.user_id)
     get_user_result = await db.execute(get_user_stmt)
     try:

app/web/html/routes/auth.py

@@ -34,6 +34,7 @@ async def login_for_access_token(
         value=token.access_token,
         httponly=True,
         secure=True,
+        samesite="lax",
     )
     return token
 

app/web/html/routes/sitemap.py

@@ -14,7 +14,7 @@
 
 
 @router.get("/sitemap.xml", response_model=None)
-@aiocache.cached(key="constant")
+@aiocache.cached(key="sitemap", ttl=3600)
 async def sitemap(request: Request, db: DBSession) -> HTMLResponse:
     """Return the sitemap page."""
     return HTMLResponse(

app/web/html/routes/users.py

@@ -31,6 +31,13 @@
 REGISTER_TEMPLATE = "users/register.html"
 
 
+def _safe_redirect(url: str | None) -> str:
+    """Return a safe same-origin redirect path, defaulting to '/'."""
+    if url and url.startswith("/") and not url.startswith("//"):
+        return url
+    return "/"
+
+
 class LoginForm(Form):
     """Form for user login page."""
 
@@ -86,7 +93,7 @@ async def login_post(
             },
             status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
         )
-    redirect_url = login_form.redirect_url.data or "/"
+    redirect_url = _safe_redirect(login_form.redirect_url.data)
 
     # Why not a RedirectResponse?
     # Because we need to set the HX-Redirect header to perform
@@ -240,7 +247,7 @@ async def register_post(
 async def logout(request: Request) -> RedirectResponse:
     """Log the user out and redirect to the home page."""
     form_data = await request.form()
-    redirect_url = str(form_data.get("next", "/"))
+    redirect_url = _safe_redirect(str(form_data.get("next", "/")))
     response = RedirectResponse(
         redirect_url,
         status_code=status.HTTP_303_SEE_OTHER,

app/web/main.py

@@ -42,7 +42,7 @@ def create_app() -> FastAPI:
         allow_methods=["*"],
         allow_headers=["*"],
     )
-    app.add_middleware(SessionMiddleware, secret_key=settings.session_secret)
+    app.add_middleware(SessionMiddleware, secret_key=settings.session_secret, max_age=86400)
 
     @app.get("/api")
     async def api_home(request: Request) -> RedirectResponse:
@@ -63,8 +63,6 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None]:  # noqa: ARG001 (unuse
         async with engine.begin() as conn:
             if settings.db_create_tables:
                 await conn.run_sync(db_models.Base.metadata.create_all)
-            else:
-                yield
     except sqlalchemy.exc.OperationalError as e:  # pragma: no cover
         err_msg = (
             "Could not connect to the database. Check the connection string."