Add verified JWT edge assertions via JWKS

.env.example

@@ -26,6 +26,14 @@ DRUKS_SECRETS_KEY=
 # Cloudflare Access: Cf-Access-Authenticated-User-Email):
 #   DRUKS_AUTH_MODE=header
 #   DRUKS_AUTH_HEADER=<your edge's identity header>
+# Or, for an edge that signs its assertions, jwt mode verifies the token in
+# DRUKS_AUTH_HEADER against the edge's JWKS:
+#   DRUKS_AUTH_MODE=jwt
+#   DRUKS_AUTH_HEADER=<your edge's assertion header>
+#   DRUKS_AUTH_JWKS_URL=https://your-edge/.well-known/jwks.json
+#   DRUKS_AUTH_JWT_ISSUER=https://your-edge
+#   DRUKS_AUTH_JWT_AUDIENCE=https://your-druks
+#   DRUKS_AUTH_JWT_IDENTITY_CLAIM=email
 DRUKS_AUTH_MODE=none
 
 # Webhook authentication and public ingress. DRUKS_WEBHOOK_HOST is used only

backend/druks/accounts/dependencies.py

@@ -4,7 +4,12 @@
 from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
 
 from druks.accounts.context import current_account_id
-from druks.accounts.exceptions import AuthConfigurationError, InvalidPatError
+from druks.accounts.exceptions import (
+    AuthConfigurationError,
+    InvalidAssertionError,
+    InvalidPatError,
+)
+from druks.accounts.jwt import verify_assertion
 from druks.accounts.models import Account, PersonalAccessToken
 
 _BEARER_CHALLENGE = 'Bearer realm="druks"'
@@ -44,14 +49,20 @@ def resolve_single_operator() -> Account | None:
     return operators[0] if operators else None
 
 
-def _resolve_operator(request: Request) -> Account | None:
-    """None only during none/zero setup. Header mode open-enrolls the edge's
-    asserted email; none mode ignores the header entirely."""
+async def _resolve_operator(request: Request) -> Account | None:
+    """None only during none/zero setup. header maps the asserted email; jwt
+    maps its verified identity claim; none ignores the header entirely."""
     settings = request.app.state.settings
     if settings.auth_mode == "none":
         return resolve_single_operator()
     values = request.headers.getlist(settings.auth_header)
-    if len(values) == 1 and (email := values[0].strip()):
+    if len(values) == 1 and (asserted := values[0].strip()):
+        if settings.auth_mode == "header":
+            return Account.get_or_create(asserted)
+        try:
+            email = await verify_assertion(asserted, settings)
+        except InvalidAssertionError as error:
+            raise HTTPException(status_code=401, detail=str(error)) from error
         return Account.get_or_create(email)
     raise HTTPException(
         status_code=401,
@@ -77,7 +88,7 @@ async def current_account(
     if "Authorization" in request.headers:
         account = resolve_pat_account(bearer)
     else:
-        account = _resolve_operator(request)
+        account = await _resolve_operator(request)
         if not account:
             raise HTTPException(
                 status_code=409,
@@ -95,7 +106,7 @@ async def current_session_account(request: Request) -> AsyncIterator[Account]:
     """The signed-in human, never a bearer — a token cannot manage
     capabilities. Identity re-asserts per request; no session state."""
     _require_no_bearer(request)
-    account = _resolve_operator(request)
+    account = await _resolve_operator(request)
     if not account:
         raise HTTPException(
             status_code=409,
@@ -112,7 +123,7 @@ async def current_session_or_setup(request: Request) -> AsyncIterator[Account |
     """The signed-in human; None during none/zero setup, where the first
     completed connection creates the operator."""
     _require_no_bearer(request)
-    account = _resolve_operator(request)
+    account = await _resolve_operator(request)
     token = current_account_id.set(account.id if account else None)
     try:
         yield account
@@ -129,7 +140,7 @@ async def current_account_or_setup(
     if "Authorization" in request.headers:
         account = resolve_pat_account(bearer)
     else:
-        account = _resolve_operator(request)
+        account = await _resolve_operator(request)
     token = current_account_id.set(account.id if account else None)
     try:
         yield account

backend/druks/accounts/exceptions.py

@@ -7,3 +7,9 @@ class AuthConfigurationError(Exception):
     """The configured auth mode cannot resolve a single operator identity —
     e.g. ``none`` mode with more than one non-system account. Refuses the
     request (and startup) instead of guessing which account is the operator."""
+
+
+class InvalidAssertionError(Exception):
+    """An edge-minted JWT assertion that fails verification — bad signature,
+    wrong issuer or audience, expired, unknown signing key, or a missing
+    identity claim. The raw token never appears in the message."""

backend/druks/accounts/jwt.py

@@ -0,0 +1,34 @@
+from functools import lru_cache
+
+from fastmcp.server.auth.providers.jwt import JWTVerifier
+
+from druks.accounts.exceptions import InvalidAssertionError
+from druks.settings import Settings
+
+# RS256 is the pinned profile — an untrusted token never chooses its own
+# algorithm.
+_ALGORITHM = "RS256"
+
+
+@lru_cache(maxsize=1)
+def _verifier(jwks_uri: str, issuer: str, audience: str) -> JWTVerifier:
+    return JWTVerifier(jwks_uri=jwks_uri, issuer=issuer, audience=audience, algorithm=_ALGORITHM)
+
+
+async def verify_assertion(token: str, settings: Settings) -> str:
+    """The verified identity claim of an edge-minted assertion, else
+    InvalidAssertionError."""
+    verifier = _verifier(
+        settings.auth_jwks_url, settings.auth_jwt_issuer, settings.auth_jwt_audience
+    )
+    access = await verifier.verify_token(token)
+    # The verifier checks signature, issuer, audience, and expiry-if-present;
+    # our contract additionally requires exp and a nonblank string identity.
+    if not access or "exp" not in access.claims:
+        raise InvalidAssertionError("Assertion rejected.")
+    claim = access.claims.get(settings.auth_jwt_identity_claim)
+    if isinstance(claim, str) and claim.strip():
+        return claim.strip()
+    raise InvalidAssertionError(
+        f"Assertion carries no usable {settings.auth_jwt_identity_claim} claim."
+    )

backend/druks/settings.py

@@ -93,11 +93,17 @@ class Settings(BaseSettings):
     # loopback-only deployments with a single operator account. ``header``: the
     # edge (exe.dev, Teleport, Cloudflare Access, …) authenticates and asserts
     # the operator's email in ``auth_header``; druks maps it to an account.
-    # Bearer personal access tokens resolve first in either mode.
-    auth_mode: Literal["none", "header"] = Field(default="none", alias="DRUKS_AUTH_MODE")
+    # ``jwt``: the edge asserts a signed JWT in ``auth_header`` instead; druks
+    # verifies it against the JWKS below and maps its identity claim. Bearer
+    # personal access tokens resolve first in every mode.
+    auth_mode: Literal["none", "header", "jwt"] = Field(default="none", alias="DRUKS_AUTH_MODE")
     # No default: the operator names their edge's header explicitly — druks
     # blesses no provider.
     auth_header: str = Field(default="", alias="DRUKS_AUTH_HEADER")
+    auth_jwks_url: str = Field(default="", alias="DRUKS_AUTH_JWKS_URL")
+    auth_jwt_issuer: str = Field(default="", alias="DRUKS_AUTH_JWT_ISSUER")
+    auth_jwt_audience: str = Field(default="", alias="DRUKS_AUTH_JWT_AUDIENCE")
+    auth_jwt_identity_claim: str = Field(default="email", alias="DRUKS_AUTH_JWT_IDENTITY_CLAIM")
 
     webhook_secret: str = Field(default="", alias="DRUKS_WEBHOOK_SECRET")
     # Public hostname webhook senders POST to (Caddy serves it; see
@@ -208,11 +214,26 @@ class Settings(BaseSettings):
     log_level: str = Field(default="INFO", alias="DRUKS_LOG_LEVEL")
 
     @model_validator(mode="after")
-    def _header_mode_names_its_header(self) -> "Settings":
-        if self.auth_mode == "header" and not self.auth_header.strip():
+    def _auth_mode_is_fully_configured(self) -> "Settings":
+        if self.auth_mode != "none" and not self.auth_header.strip():
             raise ValueError(
-                "DRUKS_AUTH_HEADER must name the edge's identity header when DRUKS_AUTH_MODE=header"
+                "DRUKS_AUTH_HEADER must name the edge's identity header "
+                f"when DRUKS_AUTH_MODE={self.auth_mode}"
             )
+        if self.auth_mode != "none" and self.auth_header.strip().lower() == "authorization":
+            # Authorization is the PAT slot and always parses bearer-first — an
+            # assertion configured there could never be read, locking everyone out.
+            raise ValueError("DRUKS_AUTH_HEADER cannot be Authorization — that slot is PAT-only")
+        if self.auth_mode == "jwt":
+            required = {
+                "DRUKS_AUTH_JWKS_URL": self.auth_jwks_url,
+                "DRUKS_AUTH_JWT_ISSUER": self.auth_jwt_issuer,
+                "DRUKS_AUTH_JWT_AUDIENCE": self.auth_jwt_audience,
+                "DRUKS_AUTH_JWT_IDENTITY_CLAIM": self.auth_jwt_identity_claim,
+            }
+            missing = [name for name, value in required.items() if not value.strip()]
+            if missing:
+                raise ValueError(f"DRUKS_AUTH_MODE=jwt requires {', '.join(missing)}")
         return self
 
     @property

backend/tests/test_identity_jwt.py

@@ -0,0 +1,124 @@
+import time
+from pathlib import Path
+
+import druks.redis
+import jwt as pyjwt
+import pytest
+from conftest import configure_app_for_test, make_settings
+from cryptography.hazmat.primitives.asymmetric import rsa
+from druks.accounts import jwt as assertion
+from druks.accounts.models import Account, PersonalAccessToken
+from fastapi.testclient import TestClient
+from fastmcp.server.auth.providers.jwt import JWTVerifier
+
+HEADER = "X-ExeDev-Email"
+KID = "edge-key-1"
+ISSUER = "https://edge.example.com"
+AUDIENCE = "druks"
+
+_PRIVATE_KEY = rsa.generate_private_key(public_exponent=65537, key_size=2048)
+_FOREIGN_KEY = rsa.generate_private_key(public_exponent=65537, key_size=2048)
+_JWK = pyjwt.algorithms.RSAAlgorithm.to_jwk(_PRIVATE_KEY.public_key(), as_dict=True)
+_JWKS = {"keys": [{**_JWK, "kid": KID}]}
+
+
+@pytest.fixture(autouse=True)
+def _serve_jwks(monkeypatch):
+    druks.redis.get_client()._data.clear()
+    assertion._verifier.cache_clear()
+
+    async def fetch_jwks(self):
+        return _JWKS
+
+    monkeypatch.setattr(JWTVerifier, "_fetch_jwks", fetch_jwks)
+    yield
+    assertion._verifier.cache_clear()
+
+
+def _token(key=_PRIVATE_KEY, **claim_overrides) -> str:
+    claims = {
+        "iss": ISSUER,
+        "aud": AUDIENCE,
+        "exp": int(time.time()) + 600,
+        "email": "op@example.com",
+        **claim_overrides,
+    }
+    claims = {name: value for name, value in claims.items() if value is not None}
+    return pyjwt.encode(claims, key, algorithm="RS256", headers={"kid": KID})
+
+
+def _jwt_client(tmp_path: Path) -> TestClient:
+    app = configure_app_for_test(
+        settings=make_settings(
+            tmp_path,
+            auth_mode="jwt",
+            auth_header=HEADER,
+            auth_jwks_url="https://edge.example.com/jwks.json",
+            auth_jwt_issuer=ISSUER,
+            auth_jwt_audience=AUDIENCE,
+        ),
+        authenticated=False,
+    )
+    return TestClient(app)
+
+
+def test_a_valid_assertion_open_enrolls_its_subject(tmp_path, db_session):
+    with _jwt_client(tmp_path) as client:
+        response = client.get("/api/auth/me", headers={HEADER: _token()})
+        assert response.status_code == 200
+        assert response.json()["account"]["username"] == "op@example.com"
+        other = client.get("/api/auth/me", headers={HEADER: _token(email="two@example.com")})
+        assert other.status_code == 200
+    usernames = {account.username for account in Account.list_non_system()}
+    assert usernames == {"op@example.com", "two@example.com"}
+
+
+@pytest.mark.parametrize(
+    "token",
+    [
+        _token(key=_FOREIGN_KEY),  # signature it can't verify
+        _token(exp=int(time.time()) - 60),  # expired
+        _token(exp=None),  # our contract requires exp even when the library allows its absence
+        _token(iss="https://impostor.example.com"),
+        _token(aud="not-druks"),
+        _token(email=None),  # missing identity claim
+        _token(email={"nested": "never"}),  # non-string identity claim
+        "not.a.jwt",
+    ],
+)
+def test_a_bad_assertion_rejects_without_enrolling(tmp_path, db_session, token):
+    with _jwt_client(tmp_path) as client:
+        response = client.get("/api/auth/me", headers={HEADER: token})
+        assert response.status_code == 401
+        # Only the failure class reaches the caller — never token material.
+        assert token.split(".")[1] not in response.json()["detail"]
+    assert not Account.list_non_system()
+
+
+def test_none_mode_multi_kid_document_serves_the_matching_key(tmp_path, db_session):
+    with _jwt_client(tmp_path) as client:
+        assert client.get("/api/auth/me", headers={HEADER: _token()}).status_code == 200
+
+
+def test_bearer_precedence_survives_jwt_mode(tmp_path, db_session):
+    agent = Account.get_or_create("agent@example.com")
+    _, token = PersonalAccessToken.create(account_id=agent.id, name="agent")
+    with _jwt_client(tmp_path) as client:
+        # A valid bearer wins over any assertion, even a garbage one.
+        response = client.get(
+            "/api/auth/me",
+            headers={"Authorization": f"Bearer {token}", HEADER: "not.a.jwt"},
+        )
+        assert response.status_code == 200
+        assert response.json()["account"]["username"] == "agent@example.com"
+        # An invalid bearer never falls through to a valid assertion.
+        bad = client.get("/api/auth/me", headers={"Authorization": "Bearer nope", HEADER: _token()})
+        assert bad.status_code == 401
+        # PAT management admits the verified assertion alone — never a bearer.
+        allowed = client.get("/api/auth/personal-tokens", headers={HEADER: _token()})
+        assert allowed.status_code == 200
+        managed = client.get(
+            "/api/auth/personal-tokens",
+            headers={"Authorization": f"Bearer {token}", HEADER: _token()},
+        )
+        assert managed.status_code == 401

backend/tests/test_settings.py

@@ -1,4 +1,5 @@
 import pytest
+from conftest import make_settings
 from druks.settings import Settings, ensure_data_dirs, load_settings
 from pydantic import ValidationError
 
@@ -19,6 +20,36 @@ def test_header_mode_requires_the_operator_to_name_the_header(tmp_path, monkeypa
         Settings(auth_mode="header", auth_header=auth_header)  # type: ignore[call-arg]
 
 
+def test_jwt_mode_requires_its_verification_targets(tmp_path):
+    complete = {
+        "auth_header": "X-Edge-Assertion",
+        "auth_jwks_url": "https://edge.example.com/jwks.json",
+        "auth_jwt_issuer": "https://edge.example.com",
+        "auth_jwt_audience": "druks",
+    }
+    settings = make_settings(tmp_path, auth_mode="jwt", **complete)
+    assert settings.auth_jwt_identity_claim == "email"
+    # Each required field, blanked in turn, refuses jwt mode.
+    for name in complete:
+        with pytest.raises(ValidationError):
+            make_settings(tmp_path, auth_mode="jwt", **{**complete, name: "  "})
+
+
+@pytest.mark.parametrize("auth_mode", ["header", "jwt"])
+def test_the_pat_slot_cannot_be_the_identity_header(tmp_path, auth_mode):
+    # Authorization always parses bearer-first, so an assertion configured
+    # there could never be read — a total lockout, refused at startup.
+    with pytest.raises(ValidationError):
+        make_settings(
+            tmp_path,
+            auth_mode=auth_mode,
+            auth_header="authorization",
+            auth_jwks_url="https://edge.example.com/jwks.json",
+            auth_jwt_issuer="https://edge.example.com",
+            auth_jwt_audience="druks",
+        )
+
+
 def test_ensure_data_dirs_provisions_skills_dir(tmp_path, monkeypatch):
     # The settings UI installs skill collections into skills_dir; if startup
     # doesn't create it, the first install's write raises OSError → opaque 500.

docs/configuration.md

@@ -35,8 +35,12 @@ caches, and the sandbox provisioning gate.
 | `DRUKS_ENDPOINT` | Browser-visible dashboard base URL used to build MCP OAuth callbacks |
 | `DRUKS_WEBHOOK_HOST` | Public webhook hostname used by `druks doctor` for its ingress probe |
 | `DRUKS_WEBHOOK_SECRET` | Shared HMAC secret used by bundled webhook integrations |
-| `DRUKS_AUTH_MODE` | `none` (default; no authentication, single operator) or `header` (edge-asserted identity) |
-| `DRUKS_AUTH_HEADER` | The trusted identity header; read by both the shipped Caddy edge and Druks. No default — header mode refuses to start without it |
+| `DRUKS_AUTH_MODE` | `none` (default; no authentication, single operator), `header` (edge-asserted identity), or `jwt` (edge-signed assertion, verified) |
+| `DRUKS_AUTH_HEADER` | The trusted identity header; read by both the shipped Caddy edge and Druks. No default — header and jwt modes refuse to start without it |
+| `DRUKS_AUTH_JWKS_URL` | `jwt` mode: where the edge publishes its signing keys |
+| `DRUKS_AUTH_JWT_ISSUER` | `jwt` mode: required `iss` claim value |
+| `DRUKS_AUTH_JWT_AUDIENCE` | `jwt` mode: required `aud` claim value |
+| `DRUKS_AUTH_JWT_IDENTITY_CLAIM` | `jwt` mode: the claim mapped to the account (default `email`) |
 
 `DRUKS_ENDPOINT` and `DRUKS_WEBHOOK_HOST` are different. The first is where an
 operator's browser reaches Druks; the second is the public ingress webhook
@@ -53,7 +57,16 @@ order:
    `DRUKS_AUTH_HEADER` value; Druks trims outer whitespace and maps it to an
    account, creating one on first sight (open enrollment — the edge decides
    who reaches Druks at all; the account column is case-insensitive).
-3. **`none` mode.** No authentication and no identity edge: Druks resolves
+3. **`jwt` mode.** The same assertion channel as `header` mode, but the value
+   is a signed JWT: Druks verifies the RS256 signature against
+   `DRUKS_AUTH_JWKS_URL` (keys cached for five minutes and refetched on
+   rotation), requires `exp`, `iss`, and `aud` to match the configured
+   issuer and audience, and maps the verified
+   `DRUKS_AUTH_JWT_IDENTITY_CLAIM` through the same open enrollment. A
+   failed verification is a 401 naming only the failure class — never the
+   token. Confirm the real edge's header name, claims, and rotation story
+   before enabling the mode; the RS256 profile is pinned, not negotiated.
+4. **`none` mode.** No authentication and no identity edge: Druks resolves
    the only non-system account. Zero accounts is the setup state — the
    dashboard onboards by connecting a harness, and the first completed
    connection creates the operator account from the provider-verified email.
@@ -64,6 +77,10 @@ Trust requirements for `header` mode: the edge must authenticate every
 dashboard request, must strip any client-supplied copy of
 `DRUKS_AUTH_HEADER` before inserting its authenticated value — a client that
 can inject the header can be anyone — and must terminate TLS and set HSTS.
+`jwt` mode keeps the same strip requirement but adds cryptographic
+provenance: a forged header value fails signature verification instead of
+becoming an identity, so a misconfigured proxy degrades to a 401 rather
+than an impersonation.
 The shipped Caddy listener is loopback HTTP behind that edge, and the Druks
 web listener itself binds loopback by default. In `none` mode there is no
 authentication at all, so the listener must stay loopback-only — never