revoke tokens issued in the same second as a password change

/auth compared the token's whole-second iat against the truncated change time, so a token issued earlier in the same second as a password change stayed valid. The change time is now rounded up, and the cookie reissued to a user who changes their own password carries that rounded-up iat so it still passes.

frigate/api/auth.py

@@ -5,6 +5,7 @@ import hashlib
 import ipaddress
 import json
 import logging
+import math
 import os
 import re
 import secrets
@@ -463,17 +464,22 @@ def validate_password_strength(password: str) -> tuple[bool, str | None]:
     return True, None
 
 
-def create_encoded_jwt(user, role, expiration, secret):
+def create_encoded_jwt(user, role, expiration, secret, iat: int | None = None):
+    if iat is None:
+        iat = int(time.time())
+
     return jwt.encode(
         {"alg": "HS256"},
-        {"sub": user, "role": role, "exp": expiration, "iat": int(time.time())},
+        {"sub": user, "role": role, "exp": expiration, "iat": iat},
         secret,
     )
 
 
 def _user_session(user: User) -> tuple[str, int | None]:
     changed_at = user.password_changed_at
-    return user.role, int(changed_at.timestamp()) if changed_at else None
+
+    # rounded up so a token issued earlier in the same second is revoked
+    return user.role, math.ceil(changed_at.timestamp()) if changed_at else None
 
 
 def load_user_sessions() -> dict[str, tuple[str, int | None]]:
@@ -1120,10 +1126,11 @@ async def update_password(
         )
 
     password_hash = hash_password(body.password, iterations=HASH_ITERATIONS)
+    changed_at = datetime.now()
     User.update(
         {
             User.password_hash: password_hash,
-            User.password_changed_at: datetime.now(),
+            User.password_changed_at: changed_at,
         }
     ).where(User.username == username).execute()
     refresh_user_session(request, username)
@@ -1137,7 +1144,11 @@ async def update_password(
 
         expiration = int(time.time()) + JWT_SESSION_LENGTH
         encoded_jwt = create_encoded_jwt(
-            username, current_role, expiration, request.app.jwt_token
+            username,
+            current_role,
+            expiration,
+            request.app.jwt_token,
+            iat=math.ceil(changed_at.timestamp()),
         )
         # Set new JWT cookie on response
         set_jwt_cookie(

frigate/test/http_api/test_http_auth_jwt_role.py

@@ -2,6 +2,7 @@
 
 import os
 import time
+from datetime import datetime
 from types import SimpleNamespace
 from unittest.mock import MagicMock, Mock, patch
 
@@ -178,9 +179,29 @@ class TestAuthJwtRole(BaseTestHttp):
             for source in ("bearer", "cookie"):
                 self.assertEqual(self._auth(client, old_token, source).status_code, 401)
 
-            resp = self._auth(client, self._token(app, "bob", "garage"))
+            changed_at = app.user_sessions["bob"][1]
+            resp = self._auth(client, self._token(app, "bob", "garage", changed_at))
             self.assertEqual(resp.status_code, 202)
 
+    def test_password_change_revokes_token_from_the_same_second(self):
+        app = self._create_app()
+        changed_at = datetime.now().replace(microsecond=500000)
+        old_token = self._token(app, "bob", "garage", iat=int(changed_at.timestamp()))
+
+        with (
+            patch("frigate.api.auth.datetime") as mock_datetime,
+            AuthTestClient(app) as client,
+        ):
+            mock_datetime.now.return_value = changed_at
+            resp = client.put(
+                "/users/bob/password",
+                json={"password": "a-brand-new-password"},
+            )
+            self.assertEqual(resp.status_code, 200)
+
+            for source in ("bearer", "cookie"):
+                self.assertEqual(self._auth(client, old_token, source).status_code, 401)
+
     def test_own_password_change_cookie_stays_valid(self):
         app = self._create_app()
         with AuthTestClient(app) as client: