Stage 381: PR #2484

api/auth.py

@@ -54,6 +54,7 @@ PUBLIC_PATHS = frozenset({
 })
 
 COOKIE_NAME = 'hermes_session'
+CSRF_HEADER_NAME = 'X-Hermes-CSRF-Token'
 
 _SESSIONS_FILE = STATE_DIR / '.sessions.json'
 
@@ -366,6 +367,37 @@ def verify_session(cookie_value: str) -> bool:
     return True
 
 
+def _session_token_from_cookie_value(cookie_value: str) -> str | None:
+    """Return the raw server-side session token from a signed cookie value."""
+    if not cookie_value or '.' not in cookie_value:
+        return None
+    token, _sig = cookie_value.rsplit('.', 1)
+    return token or None
+
+
+def csrf_token_for_session(cookie_value: str) -> str | None:
+    """Return the CSRF token bound to an authenticated WebUI session.
+
+    The browser can read this token from the authenticated shell and echoes it
+    in ``X-Hermes-CSRF-Token`` on unsafe API requests. The token is derived
+    from the HttpOnly session cookie's server-side token, so it automatically
+    rotates on login and is invalidated when the auth session expires or logs
+    out. Callers must still verify the auth session before trusting it.
+    """
+    token = _session_token_from_cookie_value(cookie_value)
+    if not token:
+        return None
+    return hmac.new(_signing_key(), f"csrf:{token}".encode(), hashlib.sha256).hexdigest()
+
+
+def verify_csrf_token(cookie_value: str, csrf_token: str) -> bool:
+    """Verify a submitted CSRF token against the authenticated session."""
+    if not cookie_value or not csrf_token or not verify_session(cookie_value):
+        return False
+    expected = csrf_token_for_session(cookie_value)
+    return bool(expected and hmac.compare_digest(str(csrf_token), expected))
+
+
 def invalidate_session(cookie_value) -> None:
     """Remove a session token."""
     if cookie_value and '.' in cookie_value:

api/routes.py

@@ -1231,13 +1231,29 @@ def _allowed_public_origins() -> set[str]:
     return result
 
 
+def _is_browser_unsafe_request(handler) -> bool:
+    """Return True when request headers identify a browser unsafe request.
+
+    Non-browser API clients, including the MCP bridge and curl-style scripts,
+    normally send no Origin/Referer and remain compatible with the existing
+    same-machine API contract. Browsers send Origin for unsafe fetch/form POSTs;
+    Referer is retained for older paths and proxies.
+    """
+    return bool(handler.headers.get("Origin") or handler.headers.get("Referer"))
+
+
+def _csrf_exempt_path(path: str) -> bool:
+    """Paths that cannot or must not carry a session CSRF token."""
+    return path in {"/api/auth/login", "/api/csp-report"}
+
+
 def _check_csrf(handler) -> bool:
-    """Reject cross-origin POST requests. Returns True if OK."""
+    """Reject cross-origin or tokenless authenticated browser unsafe requests."""
     origin = handler.headers.get("Origin", "")
     referer = handler.headers.get("Referer", "")
     host = handler.headers.get("Host", "")
-    if not origin and not referer:
-        return True  # non-browser clients (curl, agent) have no Origin
+    if not _is_browser_unsafe_request(handler):
+        return True  # non-browser clients (curl, MCP, agent) have no Origin
     target = origin or referer
     # Extract host:port from origin/referer
     m = _re.match(r"^https?://([^/]+)", target)
@@ -1246,27 +1262,39 @@ def _check_csrf(handler) -> bool:
     origin_host = m.group(1)
     origin_scheme = m.group(0).split('://')[0].lower()  # 'http' or 'https'
     origin_name, origin_port = _normalize_host_port(origin_host)
+    origin_allowed = False
     # Check against explicitly allowed public origins (env var)
     origin_value = m.group(0).rstrip('/').lower()
     if origin_value in _allowed_public_origins():
-        return True
-    # Allow same-origin: check Host, X-Forwarded-Host (reverse proxy), and
-    # X-Real-Host against the origin. Reverse proxies (Caddy, nginx) set
-    # X-Forwarded-Host to the client's original Host header.
-    allowed_hosts = [
-        h.strip()
-        for h in [
-            host,
-            handler.headers.get("X-Forwarded-Host", ""),
-            handler.headers.get("X-Real-Host", ""),
+        origin_allowed = True
+    if not origin_allowed:
+        # Allow same-origin: check Host, X-Forwarded-Host (reverse proxy), and
+        # X-Real-Host against the origin. Reverse proxies (Caddy, nginx) set
+        # X-Forwarded-Host to the client's original Host header.
+        allowed_hosts = [
+            h.strip()
+            for h in [
+                host,
+                handler.headers.get("X-Forwarded-Host", ""),
+                handler.headers.get("X-Real-Host", ""),
+            ]
+            if h.strip()
         ]
-        if h.strip()
-    ]
-    for allowed in allowed_hosts:
-        allowed_name, allowed_port = _normalize_host_port(allowed)
-        if origin_name == allowed_name and _ports_match(origin_scheme, origin_port, allowed_port):
-            return True
-    return False
+        for allowed in allowed_hosts:
+            allowed_name, allowed_port = _normalize_host_port(allowed)
+            if origin_name == allowed_name and _ports_match(origin_scheme, origin_port, allowed_port):
+                origin_allowed = True
+                break
+    if not origin_allowed:
+        return False
+
+    from api.auth import CSRF_HEADER_NAME, is_auth_enabled, parse_cookie, verify_csrf_token
+
+    if not is_auth_enabled():
+        return True
+    cookie_val = parse_cookie(handler)
+    submitted = handler.headers.get(CSRF_HEADER_NAME) or handler.headers.get("X-CSRF-Token")
+    return verify_csrf_token(cookie_val or "", submitted or "")
 
 
 def _client_ip_for_rate_limit(handler) -> str:
@@ -3324,10 +3352,22 @@ def handle_get(handler, parsed) -> bool:
             version_token = quote(WEBUI_VERSION, safe="")
             from api.extensions import inject_extension_tags
 
+            csrf_token = ""
+            try:
+                from api.auth import csrf_token_for_session, is_auth_enabled, parse_cookie, verify_session
+
+                if is_auth_enabled():
+                    cookie_val = parse_cookie(handler)
+                    if cookie_val and verify_session(cookie_val):
+                        csrf_token = csrf_token_for_session(cookie_val) or ""
+            except Exception:
+                csrf_token = ""
+
             html = (
                 _INDEX_HTML_PATH.read_text(encoding="utf-8")
                 .replace("__WEBUI_VERSION__", version_token)
                 .replace("__MAX_UPLOAD_BYTES__", str(MAX_UPLOAD_BYTES))
+                .replace("__CSRF_TOKEN_JSON__", json.dumps(csrf_token))
             )
             return t(
                 handler,
@@ -4337,10 +4377,12 @@ def handle_post(handler, parsed) -> bool:
         finally:
             if diag:
                 diag.finish()
-    # CSRF: reject cross-origin browser requests
+    # CSRF: reject cross-origin or tokenless authenticated browser requests.
+    # /api/auth/login has no authenticated session token yet, and /api/csp-report
+    # is intentionally unauthenticated for browser-generated violation reports.
     if diag:
         diag.stage("csrf")
-    if not _check_csrf(handler):
+    if not _csrf_exempt_path(parsed.path) and not _check_csrf(handler):
         try:
             return j(handler, {"error": "Cross-origin request rejected"}, status=403)
         finally:

static/index.html

@@ -26,7 +26,46 @@
 <script>(function(){try{var t=localStorage.getItem('hermes-theme')||'dark';if(t==='system')t=window.matchMedia('(prefers-color-scheme:dark)').matches?'dark':'light';var c=t==='dark'?'#141425':'#FAF7F0';document.querySelectorAll('meta[name="theme-color"]').forEach(function(m){m.setAttribute('content',c);m.removeAttribute('media');});}catch(e){}})()</script>
 <script>(function(){try{document.documentElement.dataset.workspacePanel=localStorage.getItem('hermes-webui-workspace-panel')==='open'?'open':'closed';}catch(e){document.documentElement.dataset.workspacePanel='closed';}})()</script>
 <script>(function(){try{if(localStorage.getItem('hermes-webui-sidebar-collapsed')==='1')document.documentElement.dataset.sidebarCollapsed='1';}catch(e){}})()</script>
-<script>window.__HERMES_CONFIG__={maxUploadBytes:__MAX_UPLOAD_BYTES__};</script>
+<script>window.__HERMES_CONFIG__={maxUploadBytes:__MAX_UPLOAD_BYTES__,csrfToken:__CSRF_TOKEN_JSON__};</script>
+<script>(function(){
+  var cfg=window.__HERMES_CONFIG__||{},token=cfg.csrfToken||'';
+  if(!token||!window.fetch)return;
+  var originalFetch=window.fetch.bind(window);
+  function sameOriginUnsafe(input,init){
+    var method=((init&&init.method)||(input&&input.method)||'GET').toUpperCase();
+    if(!/^(POST|PUT|PATCH|DELETE)$/.test(method))return false;
+    try{
+      var raw=(typeof input==='string'||input instanceof URL)?input:(input&&input.url)||'';
+      var url=new URL(raw,document.baseURI||location.href);
+      if(url.origin!==location.origin)return false;
+      return !/\/api\/(auth\/login|csp-report)$/.test(url.pathname);
+    }catch(_){return false;}
+  }
+  window.fetch=function(input,init){
+    if(sameOriginUnsafe(input,init)){
+      var opts=init?Object.assign({},init):{};
+      var headers=new Headers(opts.headers!==undefined?opts.headers:((input&&input.headers)||{}));
+      if(!headers.has('X-Hermes-CSRF-Token'))headers.set('X-Hermes-CSRF-Token',token);
+      opts.headers=headers;
+      return originalFetch(input,opts);
+    }
+    return originalFetch(input,init);
+  };
+  if(navigator.sendBeacon){
+    var originalBeacon=navigator.sendBeacon.bind(navigator);
+    navigator.sendBeacon=function(url,data){
+      try{
+        if(sameOriginUnsafe(url,{method:'POST'})){
+          var headers={'X-Hermes-CSRF-Token':token};
+          if(data&&data.type)headers['Content-Type']=data.type;
+          originalFetch(url,{method:'POST',credentials:'include',headers:headers,body:data,keepalive:true});
+          return true;
+        }
+      }catch(_){}
+      return originalBeacon(url,data);
+    };
+  }
+})()</script>
 <link rel="stylesheet" href="static/style.css?v=__WEBUI_VERSION__">
   <link rel="stylesheet" href="https://cdn.jsdelivr.net/npm/xterm@5.3.0/css/xterm.css" integrity="sha384-LJcOxlx9IMbNXDqJ2axpfEQKkAYbFjJfhXexLfiRJhjDU81mzgkiQq8rkV0j6dVh" crossorigin="anonymous">
   <!-- KaTeX math rendering CSS (loaded eagerly to prevent layout shift) -->

tests/test_auth_password_hash_cache.py

@@ -261,6 +261,8 @@ class TestPasswordCacheInvalidation(unittest.TestCase):
     def tearDown(self):
         if self._backup is not None:
             self._sf.write_text(self._backup, encoding='utf-8')
+        elif self._sf.exists():
+            self._sf.unlink()
         auth._invalidate_password_hash_cache()
         os.environ.pop('HERMES_WEBUI_PASSWORD', None)
 

tests/test_issue1909_csrf_token.py

@@ -0,0 +1,168 @@
+"""Regression tests for #1909 session-bound CSRF token first slice."""
+
+import hmac
+import io
+import time
+from types import SimpleNamespace
+
+import api.auth as auth
+import api.routes as routes
+
+
+def _signed_cookie(raw_token: str) -> str:
+    sig = hmac.new(auth._signing_key(), raw_token.encode(), "sha256").hexdigest()
+    auth._sessions[raw_token] = time.time() + 60
+    return f"{raw_token}.{sig}"
+
+
+class _FakeHandler:
+    def __init__(self, headers=None, body=b"{}"):
+        self.headers = headers or {}
+        self.client_address = ("127.0.0.1", 12345)
+        self.rfile = io.BytesIO(body)
+        self.wfile = io.BytesIO()
+        self.status = None
+        self.sent_headers = {}
+
+    def send_response(self, status):
+        self.status = status
+
+    def send_header(self, key, value):
+        self.sent_headers[key] = value
+
+    def end_headers(self):
+        pass
+
+
+def test_csrf_token_is_bound_to_auth_session():
+    cookie_a = _signed_cookie("a" * 64)
+    cookie_b = _signed_cookie("b" * 64)
+    try:
+        token_a = auth.csrf_token_for_session(cookie_a)
+        token_b = auth.csrf_token_for_session(cookie_b)
+
+        assert token_a and token_b and token_a != token_b
+        assert auth.verify_csrf_token(cookie_a, token_a)
+        assert not auth.verify_csrf_token(cookie_b, token_a)
+        assert not auth.verify_csrf_token(cookie_a, "not-the-token")
+    finally:
+        auth._sessions.pop("a" * 64, None)
+        auth._sessions.pop("b" * 64, None)
+
+
+def test_authenticated_same_origin_browser_post_requires_session_csrf_token(monkeypatch):
+    cookie = _signed_cookie("c" * 64)
+    token = auth.csrf_token_for_session(cookie)
+    monkeypatch.setattr(auth, "is_auth_enabled", lambda: True)
+    try:
+        base_headers = {
+            "Origin": "http://127.0.0.1:8787",
+            "Host": "127.0.0.1:8787",
+            "Cookie": f"{auth.COOKIE_NAME}={cookie}",
+        }
+        assert not routes._check_csrf(_FakeHandler(base_headers.copy()))
+
+        headers_with_token = {**base_headers, auth.CSRF_HEADER_NAME: token}
+        assert routes._check_csrf(_FakeHandler(headers_with_token))
+    finally:
+        auth._sessions.pop("c" * 64, None)
+
+
+def test_authenticated_allowed_public_origin_accepts_valid_csrf_token(monkeypatch):
+    cookie = _signed_cookie("f" * 64)
+    token = auth.csrf_token_for_session(cookie)
+    monkeypatch.setattr(auth, "is_auth_enabled", lambda: True)
+    monkeypatch.setenv("HERMES_WEBUI_ALLOWED_ORIGINS", "https://myapp.example.com:8000")
+    try:
+        headers = {
+            "Origin": "https://myapp.example.com:8000",
+            "Host": "proxy.internal",
+            "Cookie": f"{auth.COOKIE_NAME}={cookie}",
+            auth.CSRF_HEADER_NAME: token,
+        }
+        assert routes._check_csrf(_FakeHandler(headers))
+    finally:
+        auth._sessions.pop("f" * 64, None)
+
+
+def test_authenticated_reverse_proxy_same_origin_accepts_valid_csrf_token(monkeypatch):
+    cookie = _signed_cookie("g" * 64)
+    token = auth.csrf_token_for_session(cookie)
+    monkeypatch.setattr(auth, "is_auth_enabled", lambda: True)
+    try:
+        headers = {
+            "Origin": "https://example.com",
+            "Host": "127.0.0.1:8787",
+            "X-Forwarded-Host": "example.com:443",
+            "Cookie": f"{auth.COOKIE_NAME}={cookie}",
+            auth.CSRF_HEADER_NAME: token,
+        }
+        assert routes._check_csrf(_FakeHandler(headers))
+    finally:
+        auth._sessions.pop("g" * 64, None)
+
+
+def test_non_browser_mcp_style_authenticated_post_remains_compatible(monkeypatch):
+    cookie = _signed_cookie("d" * 64)
+    monkeypatch.setattr(auth, "is_auth_enabled", lambda: True)
+    try:
+        handler = _FakeHandler({"Cookie": f"{auth.COOKIE_NAME}={cookie}"})
+        assert routes._check_csrf(handler)
+    finally:
+        auth._sessions.pop("d" * 64, None)
+
+
+def test_login_route_remains_csrf_exempt(monkeypatch):
+    handler = _FakeHandler(
+        {
+            "Content-Length": "2",
+            "Content-Type": "application/json",
+            "Origin": "http://evil.example",
+            "Host": "127.0.0.1:8787",
+        }
+    )
+
+    def fail_if_called(_handler):
+        raise AssertionError("/api/auth/login must not require a pre-login CSRF token")
+
+    monkeypatch.setattr(routes, "_check_csrf", fail_if_called)
+    monkeypatch.setattr(auth, "is_auth_enabled", lambda: False)
+
+    routes.handle_post(handler, SimpleNamespace(path="/api/auth/login"))
+    assert handler.status == 200
+
+
+def test_index_shell_includes_csrf_fetch_and_sendbeacon_injection():
+    src = (routes._INDEX_HTML_PATH).read_text(encoding="utf-8")
+
+    assert "csrfToken:__CSRF_TOKEN_JSON__" in src
+    assert "X-Hermes-CSRF-Token" in src
+    assert "window.fetch=function" in src
+    assert "navigator.sendBeacon=function" in src
+    assert "auth\\/login|csp-report" in src
+
+
+def test_index_shell_injects_session_bound_csrf_token(monkeypatch):
+    cookie = _signed_cookie("e" * 64)
+    token = auth.csrf_token_for_session(cookie)
+    monkeypatch.setattr(auth, "is_auth_enabled", lambda: True)
+
+    captured = {}
+
+    def fake_t(_handler, body, *, content_type=None, **_kwargs):
+        captured["body"] = body
+        captured["content_type"] = content_type
+        return True
+
+    import api.extensions as extensions
+
+    monkeypatch.setattr(routes, "t", fake_t)
+    monkeypatch.setattr(extensions, "inject_extension_tags", lambda html: html)
+
+    try:
+        handler = _FakeHandler({"Cookie": f"{auth.COOKIE_NAME}={cookie}"})
+        assert routes.handle_get(handler, SimpleNamespace(path="/", query="")) is True
+        assert captured["content_type"] == "text/html; charset=utf-8"
+        assert f"csrfToken:{token!r}".replace("'", '"') in captured["body"]
+    finally:
+        auth._sessions.pop("e" * 64, None)

tests/test_sprint29.py

@@ -18,6 +18,7 @@ Covers:
 import importlib
 import json
 import pathlib
+import pytest
 import sys
 import time
 import urllib.error
@@ -62,6 +63,21 @@ def get_raw_with_headers(path):
 
 
 class TestCSRF:
+    @pytest.fixture(autouse=True)
+    def _disable_auth_for_origin_unit_checks(self, monkeypatch):
+        """Keep origin/port CSRF checks isolated from auth stateful tests.
+
+        These Sprint 29 cases predate session-bound CSRF tokens and exercise
+        only Origin/Referer/Host allow/deny behavior. If an earlier test leaves
+        password auth enabled in the shared pytest process, _check_csrf also
+        requires a valid session CSRF token and these origin-only assertions
+        become order-dependent. Auth-enabled token coverage lives in
+        test_issue1909_csrf_token.py.
+        """
+        import api.auth as auth
+
+        monkeypatch.setattr(auth, "is_auth_enabled", lambda: False)
+
     @staticmethod
     def _csrf_allowed(headers):
         from types import SimpleNamespace
@@ -748,8 +764,6 @@ class TestENVLock:
 
 # ── Fixture ────────────────────────────────────────────────────────────────
 
-import pytest
-
 
 @pytest.fixture(scope="module")
 def webui_server():