fix(auth): bind Keyverse account claims to local scope

CHANGELOG.d/2.12.7-keyverse-account-scope.md

@@ -0,0 +1,4 @@
+### Security
+
+- Bind production Keyverse account-derived organization, workspace, and role
+  claims to one provisioned local authorization scope before ABAC or RBAC.

backend/app/auth.py

@@ -113,6 +113,7 @@ class CurrentAccount:
     display_name: str
     preferred_locale: str | None
     corporate_entity_ids: frozenset[str]
+    process_unit_ids: frozenset[str]
     permission_codes: frozenset[str]
 
     def has_permission(self, permission_code: str) -> bool:
@@ -130,6 +131,7 @@ def _decode_access_token(token: str, settings: Settings) -> dict:
             issuer=settings.oidc_issuer,
             audience=settings.oidc_audience,
             leeway=settings.oidc_clock_skew_seconds,
+            options={"require": ["exp", "iat", "sub"]},
         )
     except HTTPException:
         raise
@@ -141,6 +143,33 @@ def _decode_access_token(token: str, settings: Settings) -> dict:
     return claims
 
 
+def _keyverse_account_claims(claims: dict) -> tuple[str, str, list[str]]:
+    """Return Keyverse's atomic account scope, rejecting ambiguous wire shapes."""
+    organization = claims.get("org")
+    workspace = claims.get("workspace")
+    roles = claims.get("role")
+    valid_roles = (
+        isinstance(roles, list)
+        and bool(roles)
+        and all(isinstance(role, str) and role.strip() for role in roles)
+        and len({role.strip() for role in roles}) == len(roles)
+    )
+    if not (
+        isinstance(organization, str)
+        and organization.strip()
+        and organization == organization.strip()
+        and isinstance(workspace, str)
+        and workspace.strip()
+        and workspace == workspace.strip()
+        and valid_roles
+    ):
+        raise HTTPException(
+            status.HTTP_401_UNAUTHORIZED,
+            "Keyverse token must contain one org, one workspace, and unique roles",
+        )
+    return organization, workspace, [role.strip() for role in roles]
+
+
 async def get_current_account(
     credentials: HTTPAuthorizationCredentials = Depends(_bearer_scheme),
     pool: asyncpg.Pool = Depends(get_pool),
@@ -149,38 +178,87 @@ async def get_current_account(
     settings = load_settings()
     claims = _decode_access_token(credentials.credentials, settings)
     subject = claims["sub"]
+    keyverse_scope = (
+        _keyverse_account_claims(claims)
+        if settings.keyverse_claim_binding_required
+        else None
+    )
 
     async with pool.acquire() as conn:
-        account_row = await conn.fetchrow(
-            "select user_account_id, display_name, preferred_locale from user_account where external_subject_id = $1",
-            subject,
-        )
+        if keyverse_scope:
+            organization, workspace, token_roles = keyverse_scope
+            account_row = await conn.fetchrow(
+                """
+                select account.user_account_id, account.display_name,
+                       account.preferred_locale, affiliation.corporate_entity_id,
+                       affiliation.process_unit_id
+                  from user_account account
+                  join account_affiliation affiliation
+                    on affiliation.user_account_id = account.user_account_id
+                  join corporate_entity entity
+                    on entity.corporate_entity_id = affiliation.corporate_entity_id
+                  join process_unit process
+                    on process.process_unit_id = affiliation.process_unit_id
+                   and process.corporate_entity_id = entity.corporate_entity_id
+                 where account.external_subject_id = $1
+                   and entity.corporate_entity_code = $2
+                   and process.process_unit_code = $3
+                """,
+                subject,
+                organization,
+                workspace,
+            )
+        else:
+            account_row = await conn.fetchrow(
+                "select user_account_id, display_name, preferred_locale from user_account where external_subject_id = $1",
+                subject,
+            )
         if account_row is None:
             raise HTTPException(
                 status.HTTP_403_FORBIDDEN,
                 "token is valid but no user_account is provisioned for this subject "
                 "(run scripts/seed_demo_data.py, or provision the account, first)",
             )
 
-        entity_rows = await conn.fetch(
-            "select corporate_entity_id from account_affiliation where user_account_id = $1",
-            account_row["user_account_id"],
-        )
-        permission_rows = await conn.fetch(
-            """
-            select distinct rp.permission_code
-            from account_role_assignment ara
-            join role_permission rp on rp.access_role_id = ara.access_role_id
-            where ara.user_account_id = $1
-            """,
-            account_row["user_account_id"],
-        )
+        if keyverse_scope:
+            entity_rows = [{"corporate_entity_id": account_row["corporate_entity_id"]}]
+            process_rows = [{"process_unit_id": account_row["process_unit_id"]}]
+            permission_rows = await conn.fetch(
+                """
+                select distinct permission.permission_code
+                  from account_role_assignment assignment
+                  join access_role role
+                    on role.access_role_id = assignment.access_role_id
+                  join role_permission permission
+                    on permission.access_role_id = assignment.access_role_id
+                 where assignment.user_account_id = $1
+                   and role.role_code = any($2::text[])
+                """,
+                account_row["user_account_id"],
+                token_roles,
+            )
+        else:
+            entity_rows = await conn.fetch(
+                "select corporate_entity_id from account_affiliation where user_account_id = $1",
+                account_row["user_account_id"],
+            )
+            process_rows = []
+            permission_rows = await conn.fetch(
+                """
+                select distinct rp.permission_code
+                from account_role_assignment ara
+                join role_permission rp on rp.access_role_id = ara.access_role_id
+                where ara.user_account_id = $1
+                """,
+                account_row["user_account_id"],
+            )
 
     return CurrentAccount(
         user_account_id=str(account_row["user_account_id"]),
         external_subject_id=subject,
         display_name=account_row["display_name"],
         preferred_locale=account_row["preferred_locale"],
         corporate_entity_ids=frozenset(str(row["corporate_entity_id"]) for row in entity_rows),
+        process_unit_ids=frozenset(str(row["process_unit_id"]) for row in process_rows),
         permission_codes=frozenset(row["permission_code"] for row in permission_rows),
     )

backend/app/config.py

@@ -37,6 +37,7 @@ class Settings:
     oidc_discovery_uri: str
     oidc_jwks_uri_override: str
     oidc_clock_skew_seconds: int
+    keyverse_claim_binding_required: bool
     # Exact browser origins allowed by CORS. Comma-separated FRONTEND_ORIGINS;
     # never a wildcard -- the backend only serves the product UI.
     frontend_origins: list[str]
@@ -122,6 +123,7 @@ def load_settings() -> Settings:
             )
         ),
         oidc_clock_skew_seconds=oidc_clock_skew_seconds,
+        keyverse_claim_binding_required=bool(keyverse_issuer),
         frontend_origins=[
             origin.strip()
             for origin in os.environ.get("FRONTEND_ORIGINS", "http://localhost:5173").split(",")

backend/app/main.py

@@ -414,10 +414,16 @@ def _rankweave_client():
 
 
 def _can_see_post(account: CurrentAccount, post: asyncpg.Record) -> bool:
-    """ABAC: public rows are visible; private rows require same-corp affiliation."""
+    """ABAC: public rows are visible; private rows require the bound local scope."""
     if post["visibility_code"] == "public":
         return True
-    return str(post["corporate_entity_id"]) in account.corporate_entity_ids
+    return (
+        str(post["corporate_entity_id"]) in account.corporate_entity_ids
+        and (
+            not account.process_unit_ids
+            or str(post.get("process_unit_id")) in account.process_unit_ids
+        )
+    )
 
 
 def _is_synthetic_demo_member(member: dict[str, Any], demo_entity_ids: set[str]) -> bool:
@@ -540,7 +546,9 @@ async def _lookup_post_labels(conn: asyncpg.Connection, rows: list[asyncpg.Recor
 
 
 async def _post_filter_options(
-    conn: asyncpg.Connection, corporate_entity_ids: frozenset[str]
+    conn: asyncpg.Connection,
+    corporate_entity_ids: frozenset[str],
+    process_unit_ids: frozenset[str],
 ) -> tuple[list[dict[str, str]], list[dict[str, str]]]:
     """Return every authorized filter value, not only values on the current page."""
     visibility_sql = f"""
@@ -552,7 +560,9 @@ async def _post_filter_options(
             on lookup.lookup_category = 'post_visibility'
            and lookup.lookup_code = post.visibility_code
          where (post.visibility_code = 'public'
-            or post.corporate_entity_id::text = any($1::text[]))
+            or (post.corporate_entity_id::text = any($1::text[])
+                and (cardinality($2::text[]) = 0
+                     or post.process_unit_id::text = any($2::text[]))))
            and {SOURCE_POST_ELIGIBILITY_SQL.format(alias='post')}
          order by display_order, code
     """
@@ -565,17 +575,19 @@ async def _post_filter_options(
             on lookup.lookup_category = 'voc_type'
            and lookup.lookup_code = post.voc_type_code
          where (post.visibility_code = 'public'
-            or post.corporate_entity_id::text = any($1::text[]))
+            or (post.corporate_entity_id::text = any($1::text[])
+                and (cardinality($2::text[]) = 0
+                     or post.process_unit_id::text = any($2::text[]))))
            and {SOURCE_POST_ELIGIBILITY_SQL.format(alias='post')}
          order by display_order, code
     """
     # Safe SQL: both query strings are closed lookup statements; entity ids remain asyncpg parameters.
     visibility_rows = await conn.fetch(  # nosemgrep: python.lang.security.audit.sqli.asyncpg-sqli.asyncpg-sqli
-        visibility_sql, list(corporate_entity_ids)
+        visibility_sql, list(corporate_entity_ids), list(process_unit_ids)
     )
     # Safe SQL: both query strings are closed lookup statements; entity ids remain asyncpg parameters.
     type_rows = await conn.fetch(  # nosemgrep: python.lang.security.audit.sqli.asyncpg-sqli.asyncpg-sqli
-        type_sql, list(corporate_entity_ids)
+        type_sql, list(corporate_entity_ids), list(process_unit_ids)
     )
     return (
         [{"code": row["code"], "label": row["label"]} for row in type_rows],
@@ -711,7 +723,10 @@ async def read_customer_master(
                   from source_post
                  where (nullif(btrim(source_customer_code), '') is not null
                         or nullif(btrim(source_customer_name), '') is not null)
-                   and (visibility_code = 'public' or corporate_entity_id = any($1::uuid[]))
+                   and (visibility_code = 'public' or (
+                        corporate_entity_id = any($1::uuid[])
+                        and (cardinality($2::uuid[]) = 0
+                             or process_unit_id = any($2::uuid[]))))
                    and {SOURCE_POST_ELIGIBILITY_SQL.format(alias='source_post')}
             ), ranked as (
                 select scoped.*,
@@ -757,6 +772,7 @@ async def read_customer_master(
              order by top_groups.post_count desc, top_groups.customer_code, top_groups.customer_name
             """,
             list(account.corporate_entity_ids),
+            list(account.process_unit_ids),
         )
         # Safe SQL: the evidence query uses only closed schema fragments; authorized entity ids are bound.
         source_author_rows = await conn.fetch(  # nosemgrep: python.lang.security.audit.sqli.asyncpg-sqli.asyncpg-sqli
@@ -777,7 +793,10 @@ async def read_customer_master(
                   join user_account author on author.user_account_id = post.author_account_id
                  where post.source_author_code is not null
                    and btrim(post.source_author_code) <> ''
-                   and (post.visibility_code = 'public' or post.corporate_entity_id = any($1::uuid[]))
+                   and (post.visibility_code = 'public' or (
+                        post.corporate_entity_id = any($1::uuid[])
+                        and (cardinality($2::uuid[]) = 0
+                             or post.process_unit_id = any($2::uuid[]))))
                    and {SOURCE_POST_ELIGIBILITY_SQL.format(alias='post')}
             ), ranked as (
                 select scoped.*,
@@ -894,6 +913,7 @@ async def read_customer_master(
              order by top_groups.post_count desc, top_groups.author_code
             """,
             list(account.corporate_entity_ids),
+            list(account.process_unit_ids),
         )
         entity_rows = await conn.fetch(
             """
@@ -1132,7 +1152,7 @@ async def list_posts(
     search_term = search.strip() if search and search.strip() else None
     async with pool.acquire() as conn:
         voc_type_options, visibility_options = await _post_filter_options(
-            conn, account.corporate_entity_ids
+            conn, account.corporate_entity_ids, account.process_unit_ids
         )
         body_search_ids: list[str] = []
         if search_term:
@@ -1173,7 +1193,7 @@ async def list_posts(
                        post.source_project_code, post.source_project_name,
                        post.source_system_code,
                        post.source_record_key,
-                       post.corporate_entity_id, post.created_at,
+                       post.corporate_entity_id, post.process_unit_id, post.created_at,
                        case
                            when $1::text is null then 0
                            when lower(coalesce(post.post_title, '')) like '%' || lower($1) || '%' then 0
@@ -1183,7 +1203,9 @@ async def list_posts(
                        count(*) over() as total_count
                   from source_post post
              where (post.visibility_code = 'public'
-                or post.corporate_entity_id::text = any($2::text[]))
+                or (post.corporate_entity_id::text = any($2::text[])
+                    and (cardinality($9::text[]) = 0
+                         or post.process_unit_id::text = any($9::text[]))))
                and {SOURCE_POST_ELIGIBILITY_SQL.format(alias="post")}
                and (
                     $1::text is null
@@ -1381,6 +1403,7 @@ async def list_posts(
             offset,
             limit,
             sort,
+            list(account.process_unit_ids),
         )
         visible = [row for row in rows if _can_see_post(account, row)]
         labels = await _lookup_post_labels(conn, visible)
@@ -1431,7 +1454,7 @@ async def read_post(
                 "source_sales_pool_code, source_sales_pool_name, "
                 "source_customer_code, source_customer_name, source_project_code, source_project_name, "
                 "source_system_code, source_record_key, "
-            "corporate_entity_id, created_at "
+            "corporate_entity_id, process_unit_id, created_at "
             f"from source_post where post_id = $1 and {SOURCE_POST_ELIGIBILITY_SQL.format(alias='source_post')}",
             post_id,
         )
@@ -1621,7 +1644,7 @@ async def _load_visible_post(
             """
             select source_post.post_id, source_post.post_title, source_post.voc_type_code,
                    source_post.visibility_code, source_post.corporate_entity_id,
-                   source_post.created_at, source_post.author_account_id,
+                   source_post.process_unit_id, source_post.created_at, source_post.author_account_id,
                    source_post.source_process_unit_code, source_post.source_author_code,
                    source_post.source_company_code, source_post.source_customer_code,
                    source_post.source_project_code, source_post.source_sales_pool_code,