reject users whose role was removed from the config

A user whose role was removed from the config fell back to viewer, which has no camera restrictions, so a restricted role removed from yaml gained access to every camera. /auth now rejects that user's tokens, and login refuses them until an admin assigns a valid role.

docker/main/rootfs/usr/local/nginx/conf/auth_location.conf

@@ -19,6 +19,7 @@ location /auth {
     # this cannot be spoofed
     proxy_set_header X-Go2rtc-Stream-Proxy $go2rtc_stream_proxy;
     proxy_set_header X-Server-Port $server_port;
+    proxy_set_header X-Forwarded-Proto $scheme;
     proxy_set_header Content-Length "";
     # Pass along auth related info
     proxy_set_header Authorization $http_authorization;

docs/docs/configuration/advanced/reference.md

@@ -117,7 +117,9 @@ auth:
   # Optional: Cookie to store the JWT token for native auth (default: shown below)
   cookie_name: frigate_token
   # Optional: Set secure flag on cookie. (default: shown below)
-  # NOTE: This should be set to True if you are using TLS
+  # NOTE: Frigate sets the flag automatically when serving TLS. Set this to True
+  # when a reverse proxy terminates TLS in front of Frigate. A proxy that connects
+  # to the TLS port must serve Frigate to browsers over HTTPS
   cookie_secure: False
   # Optional: Session length in seconds (default: shown below)
   session_length: 86400 # 24 hours

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,16 +464,47 @@ 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
+
+    # 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]]:
+    """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 is_secure_request(request: Request) -> bool:
+    # nginx sets X-Forwarded-Proto to its own scheme, replacing any client value
+    return (
+        request.app.frigate_config.auth.cookie_secure
+        or request.headers.get("x-forwarded-proto") == "https"
+    )
+
+
 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
     # same-site iframes (e.g. Home Assistant on the same host or domain), but
     # not to cross-site ones. CSRF is also mitigated by requiring a custom
@@ -750,7 +782,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,20 +807,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:
+            logger.debug("jwt user %s no longer exists", user)
+            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
 
-        # the token keeps the role it was issued with, so a role removed from
-        # 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)
+            logger.debug(
+                "jwt user %s has role %s, which is not in the config", user, role
+            )
             return fail_response
 
         current_time = int(time.time())
@@ -812,25 +850,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 +859,7 @@ def auth(request: Request):
                 JWT_COOKIE_NAME,
                 new_encoded_jwt,
                 JWT_SESSION_LENGTH,
-                JWT_COOKIE_SECURE,
+                is_secure_request(request),
             )
 
         success_response.headers["remote-user"] = user
@@ -929,7 +948,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
@@ -945,13 +963,16 @@ def login(request: Request, body: AppPostLoginBody):
 
     password_hash = db_user.password_hash
     if verify_password(password, password_hash):
-        role = getattr(db_user, "role", "viewer")
+        role = db_user.role
-        config_roles_set = set(request.app.frigate_config.auth.roles.keys())
+        if role not in request.app.frigate_config.auth.roles:
-        if role not in config_roles_set:
             logger.warning(
-                f"User {db_user.username} has an invalid role {role}, falling back to 'viewer'."
+                "Login refused for user '%s' with role '%s', which is not in the config",
+                user,
+                role,
             )
-            role = "viewer"
+            failed_logins.record(user, time.time(), known=True)
+            return JSONResponse(content={"message": "Login failed"}, status_code=401)
+
         expiration = int(time.time()) + JWT_SESSION_LENGTH
         encoded_jwt = create_encoded_jwt(user, role, expiration, request.app.jwt_token)
         response = Response("", 200)
@@ -960,7 +981,7 @@ def login(request: Request, body: AppPostLoginBody):
             JWT_COOKIE_NAME,
             encoded_jwt,
             JWT_SESSION_LENGTH,
-            JWT_COOKIE_SECURE,
+            is_secure_request(request),
         )
         # Clear admin_first_time_login flag after successful admin login so the
         # UI stops showing the first-time login documentation link.
@@ -1029,6 +1050,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 +1069,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})
 
@@ -1108,24 +1131,29 @@ 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)
 
     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
         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(
@@ -1133,7 +1161,7 @@ async def update_password(
             JWT_COOKIE_NAME,
             encoded_jwt,
             JWT_SESSION_LENGTH,
-            JWT_COOKIE_SECURE,
+            is_secure_request(request),
         )
 
     return response
@@ -1173,6 +1201,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/config/auth.py

@@ -27,7 +27,7 @@ class AuthConfig(FrigateBaseModel):
     cookie_secure: bool = Field(
         default=False,
         title="Secure cookie flag",
-        description="Set the secure flag on the auth cookie; should be true when using TLS.",
+        description="Always set the secure flag on the auth cookie. Frigate sets it automatically when serving TLS; enable this when a reverse proxy terminates TLS in front of Frigate. A proxy that connects to the TLS port must serve Frigate to browsers over HTTPS.",
     )
     session_length: int = Field(
         default=86400,

frigate/test/http_api/test_http_auth_jwt_role.py

@@ -1,22 +1,29 @@
-"""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 datetime import datetime
+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 +43,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 +75,228 @@ 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, 202)
+        self.assertEqual(resp.headers["remote-role"], "garage")
+
+    def test_role_removed_from_config_is_rejected(self):
+        User.update(role="removed_role").where(User.username == "bob").execute()
+        app = self._create_app()
+        with AuthTestClient(app) as client:
+            for source in ("bearer", "cookie"):
+                resp = self._auth(client, self._token(app, "bob", "garage"), source)
                 self.assertEqual(resp.status_code, 401)
                 self.assertNotIn("remote-role", resp.headers)
+
+    def test_login_refused_for_role_removed_from_config(self):
+        User.update(role="removed_role").where(User.username == "bob").execute()
+        with AuthTestClient(self._create_app()) as client:
+            resp = client.post("/login", json={"user": "bob", "password": PASSWORD})
+
+        self.assertEqual(resp.status_code, 401)
+        self.assertNotIn("set-cookie", resp.headers)
+
+    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)
+
+            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:
+            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"])
+
+    def _login_cookie(self, headers: dict | None = None) -> str:
+        with AuthTestClient(self._create_app()) as client:
+            resp = client.post(
+                "/login",
+                json={"user": "bob", "password": PASSWORD},
+                headers=headers or {},
+            )
+
+        self.assertEqual(resp.status_code, 200)
+        return resp.headers["set-cookie"].lower()
+
+    def test_login_cookie_is_secure_over_tls(self):
+        cookie = self._login_cookie({"x-forwarded-proto": "https"})
+        self.assertIn("; secure", cookie)
+
+    def test_login_cookie_is_not_secure_over_http(self):
+        cookie = self._login_cookie({"x-forwarded-proto": "http"})
+        self.assertNotIn("; secure", cookie)
+
+    def test_cookie_secure_config_forces_secure(self):
+        self.minimal_config["auth"]["cookie_secure"] = True
+        cookie = self._login_cookie()
+        self.assertIn("; secure", cookie)
+
+    def test_refreshed_cookie_is_secure_over_tls(self):
+        app = self._create_app()
+        token = jwt.encode(
+            {"alg": "HS256"},
+            {"sub": "bob", "exp": int(time.time()) + 60, "iat": int(time.time())},
+            app.jwt_token,
+        )
+
+        with AuthTestClient(app) as client:
+            client.cookies.set("frigate_token", token)
+            resp = client.get(
+                "/auth",
+                headers={"x-server-port": "8971", "x-forwarded-proto": "https"},
+            )
+
+        self.assertEqual(resp.status_code, 202)
+        self.assertIn("; secure", resp.headers["set-cookie"].lower())

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",

web/public/locales/en/config/global.json

@@ -40,7 +40,7 @@
     },
     "cookie_secure": {
       "label": "Secure cookie flag",
-      "description": "Set the secure flag on the auth cookie; should be true when using TLS."
+      "description": "Always set the secure flag on the auth cookie. Frigate sets it automatically when serving TLS; enable this when a reverse proxy terminates TLS in front of Frigate. A proxy that connects to the TLS port must serve Frigate to browsers over HTTPS."
     },
     "session_length": {
       "label": "Session length",