take the jwt role from the user and revoke stale sessions

/auth trusted the role claim in the token, and only checked password_changed_at when refreshing a cookie. Demoting or deleting a user, changing a password, or resetting the admin password left existing cookies and bearer tokens working with their old role. The API now keeps each user's role and password change time in memory, loads it at startup, and updates it from the user endpoints. /auth rejects unknown users and tokens issued before the last password change, and authorizes with the stored role without touching the database. Resetting the admin password now sets password_changed_at.

frigate/api/auth.py

@@ -471,6 +471,25 @@ def create_encoded_jwt(user, role, expiration, 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
+
+
+def load_user_sessions() -> dict[str, tuple[str, int | None]]:
+    """Map each username to its role and password change time for /auth."""
+    return {user.username: _user_session(user) for user in User.select()}
+
+
+def refresh_user_session(request: Request, username: str) -> None:
+    """Sync the /auth session map after a user is created, changed or deleted."""
+    user = User.get_or_none(User.username == username)
+    if user is None:
+        request.app.user_sessions.pop(username, None)
+    else:
+        request.app.user_sessions[username] = _user_session(user)
+
+
 def set_jwt_cookie(response: Response, cookie_name, encoded_jwt, max_age, secure):
     # TODO: ideally this would set secure as well, but that requires TLS
     # Starlette sets SameSite=Lax by default. The cookie is still sent to
@@ -750,7 +769,6 @@ def auth(request: Request):
     fail_response.headers["location"] = "/login"
 
     JWT_COOKIE_NAME = request.app.frigate_config.auth.cookie_name
-    JWT_COOKIE_SECURE = request.app.frigate_config.auth.cookie_secure
     JWT_REFRESH = request.app.frigate_config.auth.refresh_time
     JWT_SESSION_LENGTH = request.app.frigate_config.auth.session_length
 
@@ -776,22 +794,27 @@ def auth(request: Request):
         if "sub" not in token.claims:
             logger.debug("user not set in jwt token")
             return fail_response
-        if "role" not in token.claims:
-            logger.debug("role not set in jwt token")
-            return fail_response
         if "exp" not in token.claims:
             logger.debug("exp not set in jwt token")
             return fail_response
 
         user = token.claims.get("sub")
-        role = token.claims.get("role")
+        session = request.app.user_sessions.get(user)
-
+        if session is None:
-        # the token keeps the role it was issued with, so a role removed from
+            logger.debug("jwt user %s no longer exists", user)
-        # the config since then must send the user back through login
-        if role not in auth_config.roles:
-            logger.debug("jwt role %s is not in the config", role)
             return fail_response
 
+        role, password_changed_at = session
+        if password_changed_at is not None and (
+            int(token.claims.get("iat", 0)) < password_changed_at
+        ):
+            logger.debug("jwt token issued before password change")
+            return fail_response
+
+        # matches the fallback login applies to a role removed from the config
+        if role not in auth_config.roles:
+            role = "viewer"
+
         current_time = int(time.time())
 
         # if the jwt is expired
@@ -812,25 +835,6 @@ def auth(request: Request):
         # if the jwt cookie is expiring soon
         if jwt_source == "cookie" and expiration - JWT_REFRESH <= current_time:
             logger.debug("jwt token expiring soon, refreshing cookie")
-
-            # Check if password has been changed since token was issued
-            # If so, force re-login by rejecting the refresh
-            try:
-                user_obj = User.get_by_id(user)
-                if user_obj.password_changed_at is not None:
-                    token_iat = int(token.claims.get("iat", 0))
-                    password_changed_timestamp = int(
-                        user_obj.password_changed_at.timestamp()
-                    )
-                    if token_iat < password_changed_timestamp:
-                        logger.debug(
-                            "jwt token issued before password change, rejecting refresh"
-                        )
-                        return fail_response
-            except DoesNotExist:
-                logger.debug("user not found")
-                return fail_response
-
             new_expiration = current_time + JWT_SESSION_LENGTH
             new_encoded_jwt = create_encoded_jwt(
                 user, role, new_expiration, request.app.jwt_token
@@ -840,7 +844,7 @@ def auth(request: Request):
                 JWT_COOKIE_NAME,
                 new_encoded_jwt,
                 JWT_SESSION_LENGTH,
-                JWT_COOKIE_SECURE,
+                request.app.frigate_config.auth.cookie_secure,
             )
 
         success_response.headers["remote-user"] = user
@@ -929,7 +933,6 @@ def login(request: Request, body: AppPostLoginBody):
         )
 
     JWT_COOKIE_NAME = request.app.frigate_config.auth.cookie_name
-    JWT_COOKIE_SECURE = request.app.frigate_config.auth.cookie_secure
     JWT_SESSION_LENGTH = request.app.frigate_config.auth.session_length
     user = body.user
     password = body.password
@@ -960,7 +963,7 @@ def login(request: Request, body: AppPostLoginBody):
             JWT_COOKIE_NAME,
             encoded_jwt,
             JWT_SESSION_LENGTH,
-            JWT_COOKIE_SECURE,
+            request.app.frigate_config.auth.cookie_secure,
         )
         # Clear admin_first_time_login flag after successful admin login so the
         # UI stops showing the first-time login documentation link.
@@ -1029,6 +1032,7 @@ def create_user(
             User.notification_tokens: [],
         }
     ).execute()
+    refresh_user_session(request, body.username)
     request.app.config_publisher.publisher.publish("config/auth", None)
     return JSONResponse(content={"username": body.username})
 
@@ -1047,6 +1051,7 @@ def delete_user(request: Request, username: str):
         )
 
     User.delete_by_id(username)
+    refresh_user_session(request, username)
     request.app.config_publisher.publisher.publish("config/auth", None)
     return JSONResponse(content={"success": True})
 
@@ -1114,13 +1119,13 @@ async def update_password(
             User.password_changed_at: datetime.now(),
         }
     ).where(User.username == username).execute()
+    refresh_user_session(request, username)
 
     response = JSONResponse(content={"success": True})
 
     # If user changed their own password, issue a new JWT to keep them logged in
     if current_username == username:
         JWT_COOKIE_NAME = request.app.frigate_config.auth.cookie_name
-        JWT_COOKIE_SECURE = request.app.frigate_config.auth.cookie_secure
         JWT_SESSION_LENGTH = request.app.frigate_config.auth.session_length
 
         expiration = int(time.time()) + JWT_SESSION_LENGTH
@@ -1133,7 +1138,7 @@ async def update_password(
             JWT_COOKIE_NAME,
             encoded_jwt,
             JWT_SESSION_LENGTH,
-            JWT_COOKIE_SECURE,
+            request.app.frigate_config.auth.cookie_secure,
         )
 
     return response
@@ -1173,6 +1178,7 @@ async def update_role(
         )
 
     User.set_by_id(username, {User.role: body.role})
+    refresh_user_session(request, username)
     request.app.config_publisher.publisher.publish("config/auth", None)
     return JSONResponse(content={"success": True})
 

frigate/api/fastapi_app.py

@@ -196,7 +196,9 @@ def create_fastapi_app(
             key_bytes = str(secret).encode("utf-8")
 
         app.jwt_token = OctKey.import_key(key_bytes)
+        app.user_sessions = auth.load_user_sessions()
     else:
         app.jwt_token = None
+        app.user_sessions = {}
 
     return app

frigate/app.py

@@ -622,6 +622,7 @@ class FrigateApp:
                     username="admin",
                     role="admin",
                     password_hash=password_hash,
+                    password_changed_at=datetime.datetime.now(),
                     notification_tokens=[],
                 ).execute()
 

frigate/test/http_api/test_http_auth_jwt_role.py

@@ -1,22 +1,28 @@
-"""Tests that /auth only accepts a JWT whose role is still in the config."""
+"""Tests that /auth takes the role and revocation state from the user, not the JWT."""
 
 import os
 import time
+from types import SimpleNamespace
 from unittest.mock import MagicMock, Mock, patch
 
-from frigate.api.auth import create_encoded_jwt
+from joserfc import jwt
+
+from frigate.api.auth import create_encoded_jwt, hash_password
 from frigate.api.fastapi_app import create_fastapi_app
+from frigate.app import FrigateApp
 from frigate.config import FrigateConfig
 from frigate.config.camera.updater import CameraConfigUpdatePublisher
 from frigate.const import JWT_SECRET_ENV_VAR
-from frigate.models import Event, Recordings, ReviewSegment
+from frigate.models import Event, Recordings, ReviewSegment, User
 from frigate.test.http_api.base_http_test import AuthTestClient, BaseTestHttp
 
+PASSWORD = "correct-horse-battery"
+
 
 @patch.dict(os.environ, {JWT_SECRET_ENV_VAR: "test-secret"})
 class TestAuthJwtRole(BaseTestHttp):
     def setUp(self):
-        super().setUp(models=[Event, Recordings, ReviewSegment])
+        super().setUp(models=[Event, Recordings, ReviewSegment, User])
         self.minimal_config = {
             "mqtt": {"host": "mqtt"},
             "auth": {"enabled": True, "roles": {"garage": ["front_door"]}},
@@ -36,6 +42,19 @@ class TestAuthJwtRole(BaseTestHttp):
                 }
             },
         }
+        User.delete().execute()
+        self._insert_user("admin", "admin")
+        self._insert_user("bob", "garage")
+
+    def _insert_user(self, username: str, role: str):
+        User.insert(
+            {
+                User.username: username,
+                User.role: role,
+                User.password_hash: hash_password(PASSWORD, iterations=1),
+                User.notification_tokens: [],
+            }
+        ).execute()
 
     def _create_app(self):
         mock_publisher = Mock(spec=CameraConfigUpdatePublisher)
@@ -55,27 +74,158 @@ class TestAuthJwtRole(BaseTestHttp):
             enforce_default_admin=False,
         )
 
-    def _auth(self, app, role: str):
+    def _token(self, app, user: str, role: str, iat: int | None = None) -> str:
-        token = create_encoded_jwt("bob", role, int(time.time()) + 3600, app.jwt_token)
+        if iat is None:
-
+            return create_encoded_jwt(
-        with AuthTestClient(app) as client:
+                user, role, int(time.time()) + 3600, app.jwt_token
-            return client.get(
-                "/auth",
-                headers={
-                    "x-server-port": "8971",
-                    "authorization": f"Bearer {token}",
-                },
             )
 
+        return jwt.encode(
+            {"alg": "HS256"},
+            {"sub": user, "role": role, "exp": int(time.time()) + 3600, "iat": iat},
+            app.jwt_token,
+        )
+
+    def _auth(self, client, token: str, source: str = "bearer"):
+        headers = {"x-server-port": "8971"}
+        if source == "bearer":
+            headers["authorization"] = f"Bearer {token}"
+        else:
+            client.cookies.set("frigate_token", token)
+
+        resp = client.get("/auth", headers=headers)
+        client.cookies.clear()
+        return resp
+
     def test_configured_role_is_accepted(self):
-        resp = self._auth(self._create_app(), "garage")
+        app = self._create_app()
+        with AuthTestClient(app) as client:
+            resp = self._auth(client, self._token(app, "bob", "garage"))
 
         self.assertEqual(resp.status_code, 202)
         self.assertEqual(resp.headers["remote-user"], "bob")
         self.assertEqual(resp.headers["remote-role"], "garage")
 
-    def test_role_removed_from_config_is_rejected(self):
+    def test_role_claim_is_ignored(self):
-        resp = self._auth(self._create_app(), "removed_role")
+        app = self._create_app()
+        with AuthTestClient(app) as client:
+            resp = self._auth(client, self._token(app, "bob", "admin"))
 
-        self.assertEqual(resp.status_code, 401)
+        self.assertEqual(resp.status_code, 202)
-        self.assertNotIn("remote-role", resp.headers)
+        self.assertEqual(resp.headers["remote-role"], "garage")
+
+    def test_role_removed_from_config_falls_back_to_viewer(self):
+        User.update(role="removed_role").where(User.username == "bob").execute()
+        app = self._create_app()
+        with AuthTestClient(app) as client:
+            resp = self._auth(client, self._token(app, "bob", "removed_role"))
+
+        self.assertEqual(resp.status_code, 202)
+        self.assertEqual(resp.headers["remote-role"], "viewer")
+
+    def test_demoted_user_gets_new_role_on_next_request(self):
+        app = self._create_app()
+        token = self._token(app, "bob", "admin")
+        User.update(role="admin").where(User.username == "bob").execute()
+        app.user_sessions["bob"] = ("admin", None)
+
+        with AuthTestClient(app) as client:
+            self.assertEqual(self._auth(client, token).headers["remote-role"], "admin")
+            resp = client.put("/users/bob/role", json={"role": "viewer"})
+            self.assertEqual(resp.status_code, 200)
+
+            for source in ("bearer", "cookie"):
+                resp = self._auth(client, token, source)
+                self.assertEqual(resp.status_code, 202)
+                self.assertEqual(resp.headers["remote-role"], "viewer")
+
+    def test_deleted_user_is_rejected(self):
+        app = self._create_app()
+        token = self._token(app, "bob", "garage")
+
+        with AuthTestClient(app) as client:
+            self.assertEqual(self._auth(client, token).status_code, 202)
+            self.assertEqual(client.delete("/users/bob").status_code, 200)
+
+            for source in ("bearer", "cookie"):
+                self.assertEqual(self._auth(client, token, source).status_code, 401)
+
+    def test_created_user_is_accepted(self):
+        app = self._create_app()
+        with AuthTestClient(app) as client:
+            resp = client.post(
+                "/users",
+                json={"username": "carol", "password": PASSWORD, "role": "viewer"},
+            )
+            self.assertEqual(resp.status_code, 200)
+            resp = self._auth(client, self._token(app, "carol", "viewer"))
+
+        self.assertEqual(resp.status_code, 202)
+        self.assertEqual(resp.headers["remote-role"], "viewer")
+
+    def test_password_change_revokes_older_tokens(self):
+        app = self._create_app()
+        old_token = self._token(app, "bob", "garage", iat=int(time.time()) - 10)
+
+        with AuthTestClient(app) as client:
+            self.assertEqual(self._auth(client, old_token).status_code, 202)
+            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)
+
+            resp = self._auth(client, self._token(app, "bob", "garage"))
+            self.assertEqual(resp.status_code, 202)
+
+    def test_own_password_change_cookie_stays_valid(self):
+        app = self._create_app()
+        with AuthTestClient(app) as client:
+            resp = client.put(
+                "/users/bob/password",
+                json={"password": "a-brand-new-password", "old_password": PASSWORD},
+                headers={"remote-user": "bob", "remote-role": "garage"},
+            )
+            self.assertEqual(resp.status_code, 200)
+            new_token = resp.cookies["frigate_token"]
+
+            resp = self._auth(client, new_token, "cookie")
+
+        self.assertEqual(resp.status_code, 202)
+        self.assertEqual(resp.headers["remote-role"], "garage")
+
+    def test_admin_password_reset_revokes_older_tokens(self):
+        config = FrigateConfig(
+            **{**self.minimal_config, "auth": {"reset_admin_password": True}}
+        )
+        FrigateApp.init_auth(SimpleNamespace(config=config))
+
+        app = self._create_app()
+        old_token = self._token(app, "admin", "admin", iat=int(time.time()) - 10)
+
+        with AuthTestClient(app) as client:
+            for source in ("bearer", "cookie"):
+                self.assertEqual(self._auth(client, old_token, source).status_code, 401)
+
+    def test_auth_does_not_query_users(self):
+        app = self._create_app()
+
+        # expires inside the refresh window so the cookie refresh runs too
+        token = jwt.encode(
+            {"alg": "HS256"},
+            {"sub": "bob", "role": "garage", "exp": int(time.time()) + 60, "iat": 0},
+            app.jwt_token,
+        )
+
+        with (
+            patch.object(User, "select", side_effect=AssertionError("db query")),
+            AuthTestClient(app) as client,
+        ):
+            client.cookies.set("frigate_token", token)
+            resp = client.get("/auth", headers={"x-server-port": "8971"})
+
+        self.assertEqual(resp.status_code, 202)
+        self.assertIn("frigate_token=", resp.headers["set-cookie"])

frigate/test/http_api/test_http_go2rtc_proxy_auth.py

@@ -33,6 +33,7 @@ class TestGo2rtcProxyAuth(BaseTestHttp):
         self.minimal_config = _CONFIG
         self.app = super().create_app()
         self.app.jwt_token = OctKey.import_key(b"0" * 32)
+        self.app.user_sessions = {"admin": ("admin", None), "viewer": ("viewer", None)}
 
         get_patch = patch(
             "frigate.api.auth.requests.get",