Permissions (OPA) & authorization

Layer: Backend · authorization / OPA (cross-cutting)

Authorization is two gates that both must allow: an in-process can guard (async (resource, session) -> bool) on ops and actions, and a stateless Rego/OPA permissions service. A request proceeds only when the can guard and the OPA verdict agree.

How it works

Every check runs through fsh_lib.opa.OpaClient: a per-handler point check on get/create/update/delete, a bulk check, and a list filter that partial-evaluates data.authz.allow into a SQLAlchemy WHERE. The service is stateless — roles and bindings live in your DB and ship in the decision input on every call — and everything fails closed. You supply the two DB loaders, the opa representation, the Rego rules, and the can guards.

The config you write

Opt in with two loaders; the rest of opa.config are commented defaults.

You write — be/config/project.jsonnet (opa block):

local opa = import "be/opa/opa.libsonnet";

// Requires project.auth -- a decision is made for the session's subject.
opa: opa.config({
  // async (db) -> {role.name: {"permissions": [code, ...]}}  -> input.roles
  roles_loader: "permissions.load_roles",
  // async (subject, db) -> Sequence[fsh_lib.opa.RoleBinding]  -> input.bindings
  bindings_loader: "permissions.load_bindings",
  // opa_package: "authz",        // Rego decision package; must match the opa service
  // service_url_env: "OPA_URL",  // env var with the service base URL, read at startup
  // timeout: 5.0,                // per-decision timeout (s)
  // fail_open: false,            // false = deny on a failed call (fail-closed)
  // subject_types: ["user", "token"],  // allowed input.subject.type; lands in Subject.json's enum
}),

The two loaders bridge your DB to OPA. The decision subject is the token, but load_bindings reads the user’s live bindings, so a binding change takes effect on existing tokens immediately.

You write — be/permissions.py (trimmed):

from fsh_lib.opa import RoleBinding

async def load_roles(db: AsyncSession) -> dict[str, dict[str, object]]:
    # {role.name: {"permissions": [code, ...]}} -> input.roles.  An absent role
    # keys to an undefined permission list in Rego -- fail closed.
    rows = (await db.execute(
        select(Role.name, RolePermission.code)
        .join(RolePermission, RolePermission.role_id == Role.id),
    )).all()
    codes_by_role: dict[str, list[str]] = {}
    for name, code in rows:
        codes_by_role.setdefault(name, []).append(code)
    return {name: {"permissions": codes} for name, codes in codes_by_role.items()}

async def load_bindings(subject: Subject, db: AsyncSession) -> Sequence[RoleBinding]:
    # subject.id is the token id; bindings key off the live user
    # (subject.user["id"]).  Absent user / unknown id / no rows -> [] and
    # default-deny denies everything.
    if subject.user is None:
        return []
    # ... resolve each row's 2-D scope (all / literal / self) via RoleBinding
    # .from_row; a `self` dim with no home value drops the whole binding
    # (fail-closed), not "all" ...
    return bindings

A point check needs the row’s fields as input.resource. Pick which rep is that projection with the opa role — a normal rep; nothing on it binds it to OPA.

You write — be/config/resources/task.jsonnet (authz rep + role map):

representations: [
  // ... default / resource / list_item / comms_summary reps ...
  {
    name: "authz",
    fields: [
      { name: "completed", type: "bool" },
      { name: "project_id", type: "uuid" },
    ],
  },
],

representation_roles: {
  default: "default",
  // Sent to OPA as input.resource on every point check: the backend calls
  // to_task_authz on the loaded row, .model_dump(mode="json")s it, and ships
  // the result so a Rego rule can predicate on these fields.  Each field also
  // lands in opa/schema/resources/Task.json via the opa service's
  // `just gen sync-schemas`, so `opa check --strict` catches typo'd accesses.
  // Reps used by `opa` must be sync + fields-driven (no include_actions).
  opa: "authz",
},

The file path is the permission: authz/task/view.rego governs task:view. The routing entrypoint (_generated/entrypoint.rego, never hand-edited) parses "<resource>:<action>" and dispatches to data.authz.rule[<resource>][<action>], falling through resource → global → generic RBAC (policy/rbac.rego) — so adding a rule file routes it with no regeneration.

You write — opa/authz/task/view.rego (trimmed):

# `view` is the unified read permission: point-checked for `get`, partial-
# evaluated into the `list` WHERE.  Keep rules partial-eval-friendly (plain
# comparisons against input.resource, no constructs that block residuals) so
# `list` scopes in SQL and pagination counts stay honest.  METADATA binds the
# schemas so `opa check --strict` catches typo'd field accesses.
# METADATA
# schemas:
#   - input.resource: schema.task
#   - input.subject: schema.Subject
package authz.rule.task.view

import rego.v1
import data.rbac

default allow := false

# Baseline: a role bound to the subject carrying task:view (or *).  The generic
# engine matches subject + action + object scope (null=global, {type}=all of a
# type, {type, id}=one, {type, attrs}=attribute-scoped -> the predicate below).
allow if rbac.allow

# A completed task is publicly visible -- a plain comparison against
# input.resource, so it partial-evaluates into the list WHERE.
allow if input.resource.completed

default deny := false  # deny always wins over allow

Shell — the opa service workflow: just gen (regenerate router + sync schemas/rules), just check verify (regal lint + opa check --strict + opa test), just dev serve, just dev probe. Deep internals live in opa/README.md in the scaffold.

The frontend codegen emits

Owner-scoped resources skip the hand-written rule: set owner_field (e.g. on saved_views in project.jsonnet, requires project.opa + an opa rep that includes it) and codegen emits the whole view rule for you.

Generated — opa/_generated/rules/saved_view/view.rego (do not edit):

package authz.rule.saved_view.view

import rego.v1
import data.rbac

default allow := false

# RBAC baseline (a role carrying saved_view:view or *).
allow if rbac.allow

# Owner scoping, generated from `owner_field: owner_id`.  A plain equality, so
# it partial-evaluates into the list WHERE -- one rule governs both `get` and
# the `list` residual.
allow if input.resource.owner_id == input.subject.id

default deny := false

Permissions ride on the session (include_permissions: true on project.auth), so the FE gates with no extra fetch.

Generated — fe/src/api/permissions.ts (do not edit):

import { useSession } from "./session";

// `Permission` is discriminated over the session's own permission map, so a
// typo in `resource` or `action` is a compile error.
export function useCan(): (permission: Permission) => boolean {
  const { session } = useSession();
  const { permissions } = session;  // resource slug -> collection actions
  return useCallback((permission: Permission): boolean => {
    const entries = permissions[permission.resource] ?? [];
    return entries.some((entry) => entry.name === permission.action);
  }, [permissions]);
}

A can guard gates execution (403 on False) and per-row action visibility: a False guard drops the op from a row’s serialized actions, so the FE’s resolveObjectActions filters it out — no FE predicate needed. Object-scope ops get the row; collection-scope ops get None.

Scaffolded, safe to edit — be/tracker/guards.py:

async def can_complete_task(task: Task, _session: Session) -> bool:
    # `can` guard on the `complete` action.  A completed task's row dump omits
    # `complete` from its `actions`, so the FE never renders it.
    return not task.completed

Role bindings carry a 2-D scope (department, location), each in one of three modes — all, self (the subject’s own home dimension, resolved per request), or literal. bindingScope.ts is the single place that encoding lives, converting the picker’s sentinels to the API’s RoleBindingInput.

Scaffolded, safe to edit — fe/src/bindingScope.ts:

export function scopeValue(key: string): { mode: ScopeMode; id: string | null } {
  if (key === SELF_SCOPE) return { mode: "self", id: null };  // user's home dim, per request
  if (key === ALL_SCOPE) return { mode: "all", id: null };    // unconstrained
  return { mode: "literal", id: key };                        // a specific id
}

Gate nav items and destinations with const can = useCan() in fe/src/Shell.tsx (a UX affordance — the backend still enforces every route). Lists render through table configuration.