feat(workshop_auth): PoC C0 — cookie-path JWT auth backend + tenant i…

flocks/server/__init__.py

@@ -1 +1,22 @@
-"""Server module for HTTP API"""
+"""Server module for HTTP API.
+
+Startup hook (env-gated, zero core change):
+    If ``FLOCKS_AUTH=workshop_jwt`` is set, register the Workshop auth
+    backend + adapter on import. Keeps zero changes for deployments that
+    don't use Workshop.
+"""
+
+import os
+
+if os.getenv("FLOCKS_AUTH") == "workshop_jwt":
+    try:
+        from flocks.workshop_auth import register_workshop_auth
+
+        register_workshop_auth()
+    except Exception as _exc:  # noqa: BLE001
+        # Never crash server startup due to optional auth plugin.
+        # Operators can see the failure via logs/metrics; defaulting to
+        # LocalAuthBackend keeps the deployment usable.
+        import sys
+
+        print(f"[server] failed to register workshop auth: {_exc!r}", file=sys.stderr)

flocks/workshop_auth/__init__.py

@@ -0,0 +1,50 @@
+"""Workshop auth plugin (PoC, v1.1 route B+).
+
+Registers a JWT-based AuthBackend so the Workshop facade can call Flocks
+with ``Cookie: flocks_session=<per-team JWT>`` and get a real
+``AuthUser(tenant_ids=[...])`` — real tenant isolation via
+``flocks/contracts/access``.
+
+Production mount (env-gated, zero core changes):
+    Set ``FLOCKS_AUTH=workshop_jwt`` and Flocks will call
+    ``register_workshop_auth()`` on startup. PoC reproducer and tests
+    call it directly.
+
+    Add to ``flocks/server/__init__.py`` startup path::
+
+        if os.getenv("FLOCKS_AUTH") == "workshop_jwt":
+            from flocks.workshop_auth import register_workshop_auth
+            register_workshop_auth()
+"""
+
+import os
+
+from flocks.workshop_auth.backend import TeamJWTAuthBackend
+from flocks.workshop_auth.adapter import register_workshop_adapters
+
+__all__ = ["TeamJWTAuthBackend", "register_workshop_auth", "maybe_register_on_env"]
+
+
+def register_workshop_auth() -> None:
+    """Register the auth backend AND the request-context adapter.
+
+    Idempotent: re-registration is safe — Flocks' register_backend
+    overwrites the previous backend and register_auth_context_adapter
+    is dict-assignment.
+    """
+    from flocks.auth.service import AuthService
+
+    AuthService.register_backend(TeamJWTAuthBackend)
+    register_workshop_adapters()
+
+
+def maybe_register_on_env(env_var: str = "FLOCKS_AUTH", expected: str = "workshop_jwt") -> bool:
+    """Conditionally register based on env. Returns True if registered.
+
+    Designed to be called from the Flocks startup hook. Keeps zero changes
+    to the core server when the env is not set.
+    """
+    if os.getenv(env_var) == expected:
+        register_workshop_auth()
+        return True
+    return False

flocks/workshop_auth/adapter.py

@@ -0,0 +1,99 @@
+"""Workshop permission adapter (PoC, v1.1 route B+).
+
+Injects Workshop-side RBAC permissions into ``request.state.extension_context``
+after auth resolution, so any downstream code can read
+``request.state.extension_context["wk_permissions"]``.
+
+Mount:
+    The adapter is registered automatically when
+    ``flocks.workshop_auth.register_workshop_auth()`` runs. See ``__init__.py``.
+
+Real signature (verified against ``flocks/server/auth.py`` L28-33)::
+
+    AuthContextAdapter = Callable[
+        [HTTPConnection, AuthUser | None],
+        Awaitable[Mapping[str, Any] | None],
+    ]
+
+We DO NOT re-verify the JWT here — the auth backend has already decoded it
+on the same cookie in ``get_user_by_session_id``. We only unwrap the claims
+to surface permissions for RBAC checks. Re-verification would cost a JWKS
+round-trip per request.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Mapping
+from typing import Any
+
+from starlette.requests import HTTPConnection
+
+from flocks.auth.context import AuthUser
+
+_PERMISSION_KEY = "wk_permissions"
+_USER_ID_KEY = "wk_user_id"
+_TENANT_IDS_KEY = "wk_tenant_ids"
+
+
+async def inject_workshop_context(
+    request: HTTPConnection,
+    user: AuthUser | None,
+) -> Mapping[str, Any] | None:
+    """Surface Workshop claims into request.state.extension_context.
+
+    Returns None when:
+      - the request is not a Workshop-served request (no flocks_session cookie
+        OR the cookie is not a JWT we issued)
+      - the user object is missing tenant_ids (e.g. LocalAuthBackend admin)
+
+    Otherwise returns a dict that ``apply_auth_for_request`` will merge into
+    ``request.state.extension_context``.
+    """
+    if user is None:
+        return None
+    if user.role != "member":
+        # admin / service-user sessions do not carry tenant scope
+        # (PolicyContextResolver returns empty context for them, §10-C0 ③)
+        return None
+
+    cookie = request.cookies.get("flocks_session")
+    if not cookie:
+        return None
+
+    claims = _safe_decode(cookie)
+    if claims is None:
+        return None
+
+    permissions = claims.get("permissions") or []
+    teams = claims.get("teams") or ()
+    return {
+        _PERMISSION_KEY: list(permissions),
+        _USER_ID_KEY: str(claims.get("sub", user.id)),
+        _TENANT_IDS_KEY: tuple(teams),
+    }
+
+
+def _safe_decode(token: str) -> dict | None:
+    """Decode WITHOUT re-verifying signature.
+
+    The auth backend decoded & verified this exact token moments ago in
+    ``get_user_by_session_id``. Re-verification would require a JWKS fetch
+    on every request and gain no security (the cookie is HttpOnly + same
+    origin). We only trust the value because the backend just trusted it.
+
+    For paranoia in untrusted deployments, swap this for a JWKS verify —
+    the perf cost is one cached HTTP call per kid.
+    """
+    try:
+        import jwt as pyjwt
+
+        return pyjwt.decode(token, options={"verify_signature": False})
+    except Exception:
+        return None
+
+
+def register_workshop_adapters() -> None:
+    """Register the Workshop context adapter under name ``wk_rbac``."""
+    from flocks.server.auth import register_auth_context_adapter
+
+    register_auth_context_adapter("wk_rbac", inject_workshop_context)

flocks/workshop_auth/backend.py

@@ -0,0 +1,183 @@
+"""TeamJWTAuthBackend — Workshop-issued JWT → Flocks AuthUser (PoC).
+
+Mount path (verified against flocks/server/auth.py::_apply_auth_for_request):
+the session-cookie branch is checked FIRST, before browser detection, and it
+calls ``AuthService.get_user_by_session_id(session_id)`` which delegates to
+the registered backend. The Workshop facade therefore sends::
+
+    Cookie: flocks_session=<per-team JWT>
+
+and this backend decodes the JWT and returns a LocalUser whose
+``to_auth_user()`` carries ``tenant_ids`` — the field consumed by
+``flocks/contracts/access`` PolicyContext for tenant-level filtering.
+
+PoC simplification: HS256 with a shared secret from env WORKSHOP_JWT_SECRET.
+Production: RS256/ES256 via JWKS (WORKSHOP_JWKS_URL) — see design doc §6.2.
+"""
+
+from __future__ import annotations
+
+import os
+from datetime import UTC, datetime
+from typing import Optional, Tuple
+
+import jwt as pyjwt
+
+from flocks.auth.context import AuthUser
+from flocks.utils.log import Log
+
+log = Log.create(service="workshop_auth")
+
+# ---------------------------------------------------------------------------
+# Algorithm selection — PoC defaults to HS256 for the lightweight standalone
+# reproducer; production should set WORKSHOP_JWT_ALG=RS256 + WORKSHOP_JWKS_URL.
+# ---------------------------------------------------------------------------
+_ALG_ENV = "WORKSHOP_JWT_ALG"
+_JWKS_URL_ENV = "WORKSHOP_JWKS_URL"
+_HS_SECRET_ENV = "WORKSHOP_JWT_SECRET"
+
+
+def _algorithm() -> str:
+    return os.getenv(_ALG_ENV, "HS256").upper()
+
+
+async def _decode(token: str) -> dict:
+    alg = _algorithm()
+    audience = "flocks"
+    options = {"verify_aud": True, "verify_exp": True}
+    if alg == "HS256":
+        secret = os.getenv(_HS_SECRET_ENV, "")
+        if not secret:
+            raise RuntimeError(f"env {_HS_SECRET_ENV} is required for HS256")
+        payload = pyjwt.decode(token, secret, algorithms=["HS256"], audience=audience, options=options)
+    elif alg in ("RS256", "RS384", "RS512", "ES256", "ES384"):
+        from flocks.workshop_auth.client import public_key_for
+
+        jwks_url = os.getenv(_JWKS_URL_ENV, "")
+        if not jwks_url:
+            raise RuntimeError(f"env {_JWKS_URL_ENV} is required for {alg}")
+        try:
+            unverified_header = pyjwt.get_unverified_header(token)
+        except pyjwt.DecodeError as e:
+            raise pyjwt.InvalidTokenError(f"malformed JWT header: {e}") from e
+        kid = unverified_header.get("kid")
+        key = await public_key_for(kid, jwks_url)
+        if key is None:
+            raise pyjwt.InvalidTokenError("unable to resolve JWKS key for token")
+        payload = pyjwt.decode(token, key, algorithms=[alg], audience=audience, options=options)
+    else:
+        raise RuntimeError(f"unsupported WORKSHOP_JWT_ALG={alg!r}")
+    if payload.get("iss") != "ai-agent-workshop":
+        raise pyjwt.InvalidIssuerError("unexpected issuer")
+    return payload
+
+
+
+
+
+class TeamJWTAuthBackend:
+    """AuthBackend implementation backed by Workshop-issued JWTs.
+
+    User authority lives in the Workshop facade; Flocks only consumes the
+    identity/tenant claims embedded in the short-lived token.
+    """
+
+    # ---- lifecycle -------------------------------------------------------
+
+    @classmethod
+    async def init(cls) -> None:
+        return None  # no Flocks-side tables; RBAC lives in Workshop
+
+    @classmethod
+    async def has_users(cls) -> bool:
+        return True  # users are managed by Workshop; never blocks bootstrap
+
+    @classmethod
+    async def get_bootstrap_status(cls) -> dict:
+        return {"admin": True, "member": True}
+
+    @classmethod
+    async def bootstrap_admin(cls, username: str, password: str):
+        raise NotImplementedError("users managed by Workshop")
+
+    # ---- session resolution (the path the cookie branch uses) ------------
+
+    @classmethod
+    async def get_user_by_session_id(cls, session_id: str):
+        # Lazy import to avoid a cycle: LocalUser is defined in service.py.
+        from flocks.auth.service import LocalUser
+
+        # 集成契约: token 无效时返回 None(框架语义 → 401 "登录已过期"),
+        # 而非让 pyjwt 异常裸穿(全局异常处理器会包成 500, 破坏客户端
+        # 重试语义)。真实服务联调(2026-09-03)发现并修正。
+        try:
+            payload = await _decode(session_id)
+        except Exception:
+            return None
+        return LocalUser(
+            id=payload["sub"],
+            username=payload.get("username", payload["sub"]),
+            role=payload.get("flocks_role", "member"),
+            status="active",
+            must_reset_password=False,
+            tenant_ids=tuple(payload.get("teams", ())),
+            asset_groups=tuple(payload.get("asset_groups", ())),
+            created_at=payload.get("iat_iso", datetime.now(UTC).isoformat()),
+            updated_at=payload.get("iat_iso", datetime.now(UTC).isoformat()),
+            last_login_at=None,
+        )
+
+    @classmethod
+    async def revoke_session(cls, session_id: str) -> None:
+        # JWTs are short-lived and stateless; nothing to revoke Flocks-side.
+        return None
+
+    # ---- not used by the facade integration; kept for Protocol parity ----
+
+    @classmethod
+    async def login(cls, username: str, password: str, *, persist: bool = True):
+        raise NotImplementedError("login via Workshop facade only")
+
+    @classmethod
+    async def get_user_by_id(cls, user_id: str):
+        raise NotImplementedError("user directory lives in Workshop")
+
+    @classmethod
+    async def get_user_by_username(cls, username: str):
+        raise NotImplementedError("user directory lives in Workshop")
+
+    @classmethod
+    async def list_users(cls):
+        return []
+
+    @classmethod
+    async def create_user(cls, *, username: str, password: str, role: str):
+        raise NotImplementedError("users managed by Workshop")
+
+    @classmethod
+    async def update_user_role(cls, *, target_user_id: str, new_role: str):
+        raise NotImplementedError("users managed by Workshop")
+
+    @classmethod
+    async def delete_user(cls, *, target_user_id: str) -> None:
+        raise NotImplementedError("users managed by Workshop")
+
+    @classmethod
+    async def change_password(cls, user: AuthUser, *, current_password: str, new_password: str) -> None:
+        raise NotImplementedError("passwords managed by Workshop")
+
+    @classmethod
+    async def set_password(cls, **kwargs) -> None:
+        raise NotImplementedError("passwords managed by Workshop")
+
+    @classmethod
+    async def generate_admin_temp_password(cls, *, username: str = "admin") -> str:
+        raise NotImplementedError("users managed by Workshop")
+
+    @classmethod
+    async def reassign_orphan_sessions(cls, admin_user_id: str, *, dry_run: bool = False):
+        return {"reassigned": 0}
+
+    @classmethod
+    async def migrate_legacy_sessions_to_admin(cls, admin_user_id: str) -> None:
+        return None