Skip to content

IAM Engine

Verified against the code · 19d408be · 2026-08-25

Re-verified 2026-09-02 after hardening 0.2 (#520) + 0.9 (#531): ALL relation write-paths are node-scoped gated — preview/commit + delete + heal-uuid via _iam_reject_if_locked on the source node, and document-attach via _gate_item_write. The KM source documents the two-helper model (mutations router vs Excel-import) incl. both 403 bodies. No remaining gaps in relation write authorization; IamGuard unification is trin 3.2.

Layer: domain

Overview — What IAM Governs

The IAM Engine controls which nodes in a domain a user can read, write, or administer. It operates on a graph that is organized as one or more trees, and it enforces access at the subtree level: an administrator grants a user a role on a root node, and that grant propagates downward through all descendants of that node.

The engine is composed of seven focused modules:

Module Responsibility
iam/__init__.py Pure evaluation core — no I/O, no HTTP
iam/config.py IAM_ENABLED feature flag
iam/admin.py SQLite persistence and rule lifecycle (CRUD + load_access_rules)
iam/enforcement.py Bulk and single-node access checks used on read paths
iam/tree.py Builds the IAM-filtered tree response
iam/relations.py Builds the IAM-filtered relation list
iam/export.py Builds an IAM-filtered export snapshot
iam/preview.py Evaluates a rule's effective scope without persisting it
iam/overview.py Composes per-rule previews and a merged total-access view

IAM is an opt-in feature. All enforcement is guarded by is_iam_enabled(). When the flag is off every caller treats the user as having full access, so the engine adds zero overhead to deployments that do not need it.


Permission Model

Roles

Three roles are defined. Each role is a strict superset of the one below it.

Role Permitted actions
viewer read
editor read, write
admin read, write, admin

The mapping is a frozenset constant _ROLE_PERMISSIONS defined in iam/__init__.py and is the single source of truth for what each role may do.

Data model for assignments — AccessRule

@dataclass(frozen=True)
class AccessRule:
    subject:      str           # user_id
    role:         str           # "viewer" | "editor" | "admin"
    domain:       str           # e.g. "product"
    subtree_root: Optional[str] # None = full-instance grant
    granted_by:   str           # admin user_id who created this rule
    granted_at:   str           # ISO 8601 timestamp

subtree_root=None is a full-instance grant that matches every node in the domain and is treated as the lowest-priority match (depth -1) during evaluation. When subtree_root is set to a node ID, the grant covers that node and every descendant.

Rules are immutable frozen dataclasses in memory. Persistence is in a SQLite table access_rules (columns: id, subject, role, domain, subtree_root, granted_by, granted_at).

AuthContext

@dataclass(frozen=True)
class AuthContext:
    user_id: str
    email:   str = ""   # display/logging only; never evaluated

AuthContext is the sole auth input for all enforcement calls. It is produced by backend/auth.py::resolve_identity() and passed through the FastAPI dependency chain.

Super Admin

The sentinel subject __super__ represents the instance's built-in super-admin account. It bypasses all IAM enforcement silently. create_rule, list_rules, and get_rule all reject or omit __super__ so the account is never visible in the admin UI.


Core Operations

is_ancestor_or_equal

def is_ancestor_or_equal(
    candidate_root: str,
    target: str,
    parent_of: dict[str, str],
) -> bool:
Param Type Description
candidate_root str Node being tested as an ancestor
target str Node whose ancestry is walked
parent_of dict[str, str] Maps child_id → parent_id; roots are absent

Returns True if candidate_root equals target or is an ancestor of target. Walks up the parent_of chain until the root is found or the chain ends without a match. This is the tree traversal primitive that all subtree-scope checks delegate to.


get_effective_role

def get_effective_role(
    user_id:   str,
    domain:    str,
    node_id:   str,
    rules:     list[AccessRule],
    parent_of: dict[str, str],
) -> Optional[str]:
Param Type Description
user_id str Subject being evaluated
domain str Domain to filter rules by
node_id str Node to evaluate access on
rules list[AccessRule] All loaded rules for the instance
parent_of dict[str, str] Tree topology

Returns the role string ("viewer", "editor", "admin") or None if no rule matches.

Algorithm: 1. Filter rules to those whose subject == user_id and domain == domain. 2. For each matching rule, check whether subtree_root is an ancestor-or-equal of node_id (using is_ancestor_or_equal). Full-instance rules (subtree_root=None) always match at depth -1. 3. Among all matching rules collect (depth, role) pairs where depth is the number of hops from subtree_root to the root of the tree. 4. Return the role from the pair with the greatest depth (most-specific wins).

Most-specificity ensures that a narrower subtree-editor grant overrides a broader subtree-viewer grant on the same node.


evaluate_access

def evaluate_access(
    user_id:   str,
    domain:    str,
    node_id:   str,
    action:    str,
    rules:     list[AccessRule],
    parent_of: dict[str, str],
) -> bool:
Param Type Description
action str "read", "write", or "admin"

Returns True if the user's effective role permits action. Default-deny: returns False when no rule matches (contract IAM-CCR-3).


access_granted_for_context

def access_granted_for_context(
    auth:      AuthContext,
    domain:    str,
    node_id:   str,
    action:    str,
    rules:     list[AccessRule],
    parent_of: dict[str, str],
) -> bool:

The public entry point that accepts AuthContext instead of a raw user_id. All enforcement in iam/enforcement.py routes through this function. It simply unpacks auth.user_id and calls evaluate_access.


filter_nodes_by_access

def filter_nodes_by_access(
    all_nodes:   dict[str, dict],
    user_id:     str,
    domain:      str,
    rules:       list[AccessRule],
    parent_of:   dict[str, str],
    children_of: dict[str, list[str]],
) -> dict[str, dict]:
Param Type Description
all_nodes dict[str, dict] Full node map keyed by node_id

Returns the subset of all_nodes for which access_granted_for_context(..., action="read") returns True. Used by the /nodes listing endpoint when IAM is enabled.


is_node_accessible

def is_node_accessible(
    node_id:     str,
    user_id:     str,
    domain:      str,
    rules:       list[AccessRule],
    parent_of:   dict[str, str],
    children_of: dict[str, list[str]],
) -> bool:

Single-node read check. Delegates to access_granted_for_context with action="read".


has_write_access

def has_write_access(
    node_id:     str,
    user_id:     str,
    domain:      str,
    rules:       list[AccessRule],
    parent_of:   dict[str, str],
    children_of: dict[str, list[str]],
) -> bool:

Single-node write check. Delegates to access_granted_for_context with action="write". Called by the write gates — _iam_reject_if_locked in the mutations router (node validate/preview/node-update + relation preview/commit) and _require_iam_write in server.py (Excel-import endpoints).


build_visible_tree

def build_visible_tree(
    user_id:     str,
    domain:      str,
    rules:       list[AccessRule],
    all_nodes:   dict[str, dict],
    parent_of:   dict[str, str],
    children_of: dict[str, list[str]],
    roots:       list[str],
) -> list[dict]:
Param Type Description
roots list[str] Root node IDs for this domain

Returns a list of root-level node dicts each containing embedded children. Two node shapes are produced:

Allowed node — full data, restricted=False, effective_role set:

{
  "id": "...",
  "label": "...",
  "restricted": false,
  "effective_role": "editor",
  "class": "...",
  "productType": null,
  "properties": {},
  "children": [...]
}

Restricted path node — navigation skeleton, restricted=True, no class, only description property preserved:

{
  "id": "...",
  "label": "...",
  "restricted": true,
  "effective_role": null,
  "class": null,
  "properties": {"description": "..."},
  "children": [...]
}

Path ancestors (nodes that are not in the user's allowed set but have at least one accessible descendant) appear as restricted nodes so the frontend can render a navigable path to the accessible subtree. Branches with no accessible content are omitted entirely.


build_visible_relations

def build_visible_relations(
    user_id:     str,
    domain:      str,
    rules:       list[AccessRule],
    relations:   list[dict],
    all_nodes:   dict[str, dict],
    parent_of:   dict[str, str],
    children_of: dict[str, list[str]],
) -> list[dict]:

Returns relations whose from node is in the user's allowed set. For each such relation the target field is enriched:

  • Target in allowed set: full data, restricted=False.
  • Target exists but is out of scope: {"id": "...", "label": "Restricted", "restricted": true}.
  • Cross-domain target (to=None): {"id": null, "label": "Restricted", "restricted": true} — always restricted regardless of role; external_domain and external_ref are preserved at the relation level.

build_export

def build_export(
    user_id:     str,
    domain:      str,
    rules:       list[AccessRule],
    all_nodes:   dict[str, dict],
    parent_of:   dict[str, str],
    children_of: dict[str, list[str]],
    relations:   list[dict],
) -> dict:

Returns:

{
  "nodes": [{"id", "label", "class", "properties"}, ...],
  "relations": [...],
  "meta": {
    "domain": "...",
    "user_id": "...",
    "exported_at": "2026-...",
    "node_count": 42,
    "relation_count": 7
  }
}

Export differs from the tree response: path ancestors are not included. Only nodes in the allowed set appear in nodes. Relations follow the same visibility rules as build_visible_relations.


preview_access_scope

def preview_access_scope(
    subject:      str,
    domain:       str,
    rules:        list[AccessRule],
    all_nodes:    dict[str, dict],
    parent_of:    dict[str, str],
    children_of:  dict[str, list[str]],
    relations:    list[dict],
    subtree_root: Optional[str] = None,
    include_ids:  bool = False,
) -> dict:
Param Type Description
subtree_root Optional[str] The root being previewed; used to look up the subtree label
include_ids bool When True, includes the full node_ids list in the scope dict

Calls build_export once and derives counts, a 50-node preview sorted by tree depth, and the most-specific role. Zero new policy logic — all filtering is delegated. Used by both the hypothetical-rule preview endpoint and the existing-rule preview endpoint.


build_user_access_overview

def build_user_access_overview(
    subject:     str,
    domain:      str,
    rule_dicts:  list[dict],
    all_nodes:   dict[str, dict],
    parent_of:   dict[str, str],
    children_of: dict[str, list[str]],
    relations:   list[dict],
    include_ids: bool = False,
) -> dict:

Composes per-rule isolation previews (each rule evaluated in isolation via preview_access_scope) with a merged total-access view (via build_export). Returns:

{
  "subject": "...",
  "rules": [
    {
      "id": "...", "role": "...", "domain": "...",
      "subtree_root": "...", "subtree_root_label": "...",
      "granted_by": "...", "granted_at": "...",
      "preview": {"node_count": 12, "relation_count": 3, "nodes_preview": [...]}
    }
  ],
  "total_access": {
    "node_count": 20, "relation_count": 5,
    "domains": ["product"], "nodes_preview": [...]
  },
  "overlap_hint": {
    "has_overlap": true,
    "sum_of_rule_node_counts": 25,
    "note": "..."
  }
}

When include_ids=True the response additionally includes node_ids on each rule preview, on total_access, and expands overlap_hint with overlapping_node_count and overlapping_nodes.


Persistence — admin.py

init_db

def init_db(db_path: str) -> None:

Creates the access_rules table with WAL journal mode if it does not already exist. Called once at server startup.

create_rule

def create_rule(
    db_path:        str,
    *,
    subject:        str,
    role:           str,
    domain:         str,
    subtree_root:   Optional[str],
    granted_by:     str,
    known_domains:  Optional[set[str]] = None,
    valid_node_ids: Optional[set[str]] = None,
) -> dict:
Param Type Description
known_domains Optional[set[str]] If provided, domain must be a member
valid_node_ids Optional[set[str]] If provided and subtree_root is set, subtree_root must be a member

Validates subject (not __super__), role, domain, and subtree_root, then inserts the row with a UUID v4 id and a UTC granted_at timestamp. Returns the full rule dict.

list_rules

def list_rules(
    db_path: str,
    subject: Optional[str] = None,
) -> list[dict]:

Returns all rules, optionally filtered by subject. Super Admin rules are always excluded.

get_rule

def get_rule(db_path: str, rule_id: str) -> Optional[dict]:

Returns a single rule by rule_id, or None if not found or if subject == __super__.

update_rule

def update_rule(
    db_path:        str,
    rule_id:        str,
    *,
    role:           Optional[str] = None,
    subtree_root:   object = _SENTINEL,
    known_domains:  Optional[set[str]] = None,
    valid_node_ids: Optional[set[str]] = None,
) -> Optional[dict]:

Updates role and/or subtree_root on an existing rule. The _SENTINEL default distinguishes "caller did not pass subtree_root" from "caller explicitly passed subtree_root=None" (the latter is a valid full-instance grant). Returns the updated dict, or None if the rule does not exist.

delete_rule

def delete_rule(db_path: str, rule_id: str) -> bool:

Deletes a rule by ID. Returns True if the row was deleted, False if it was not found.

delete_rules_by_subject

def delete_rules_by_subject(db_path: str, subject: str) -> int:

Deletes every rule whose subject matches. Called by admin_delete_user to keep IAM rules consistent with the user store. Returns the number of rows deleted.

load_access_rules

def load_access_rules(
    db_path: str,
    domain:  Optional[str] = None,
) -> list[AccessRule]:

The bridge between SQLite and IAM core. Reads rows and constructs AccessRule dataclass instances. The optional domain filter reduces I/O for single-domain evaluations. Every enforcement call in server.py calls load_access_rules at request time.


Rule Evaluation

How rules are checked at request time

At the start of any request that needs IAM enforcement, server.py calls _get_iam_rules():

def _get_iam_rules() -> list:
    return load_access_rules(_IAM_RULES_DB)

This loads all rules fresh from SQLite. The rules are then passed — together with the graph topology (parent_of, children_of) from the effective graph — into whichever IAM function the endpoint needs.

The evaluation is always stateless and pure. No rules are cached between requests (cache invalidation happens for the tree shape cache, not for IAM rule data).

Which endpoints enforce IAM

Read paths — filtered silently

  • GET /tree — calls build_visible_tree; the response is pre-filtered before it leaves the server.
  • GET /nodes — calls filter_nodes_by_access; only accessible nodes are returned.
  • GET /node/{node_id} — calls is_node_accessible; returns 403 if the node is not in scope. Also gates cross-domain relation pills through _has_domain_read_access.
  • GET /relations — calls build_visible_relations.
  • GET /export — calls build_export.

Write paths — hard gate before mutation

Two helpers enforce the node-scoped write gate; both delegate to has_write_access:

  • _iam_reject_if_locked(auth, target_id) — the mutations router's gate (backend/api/mutations_router.py). Applied to node validate/preview/node-update and, since hardening 0.2 (#512), to relation preview and commit (gated on the relation's source node — a relation is owned by its source, so creating one is a write to the source subtree; the commit gate runs on the server-stored preview command, so sourceId cannot be swapped). Raises HTTPException(403, detail="Insufficient access"). No-ops when IAM is disabled or target_id is empty.
  • _require_iam_write(user_id, node_id) — the server.py helper used by the Excel-import endpoints (mutations_excel_router.py, BL-IAM-008):
def _require_iam_write(user_id: str, node_id: str) -> None:
    if not is_iam_enabled() or not node_id:
        return
    rules = _get_iam_rules()
    data  = loader._load()
    if not has_write_access(
        node_id, user_id, _THIS_DOMAIN,
        rules, data["parent_of"], data["children_of"],
    ):
        raise HTTPException(status_code=403, detail="Insufficient write access on target")

Affected endpoints: POST /mutations/node/validate, POST /mutations/node/preview, POST /mutations/node-update/preview, POST /mutations/relation/preview, POST /mutations/relation/commit (via _iam_reject_if_locked), and all /mutations/import/* endpoints (via _require_iam_write). The node checks target the parent node (the node that will own the change), so a viewer on a subtree cannot add, move, or delete nodes within it; the relation checks target the source node.

All relation write-paths are node-scoped gated (hardening 0.9, #531): DELETE /mutations/relation/{rel_uuid} and PATCH /mutations/relation/heal-uuid call _iam_reject_if_locked on the relation's source node (delete resolves the source from the overlay file, else the effective graph; the gate is applied only when IAM is enabled so it never precedes the engine's canonical 404/already-deleted on non-IAM deployments). The document-attach path (routes/documents.py::_attach_document) was already gated by _gate_item_write(targetNodeId) before every call site (single attach + mass-assign). Unifying both helpers into one IamGuard dependency remains hardening trin 3.2.

Cross-domain proxy gate

Proxy endpoints that forward requests to secondary backends use two FastAPI dependencies:

  • _gate_proxy_domain — requires any IAM rule for the proxied domain (read floor).
  • _gate_proxy_write — requires an editor or admin rule for the proxied domain.

Both are no-ops when IAM is disabled or when the current instance is not a primary.

Admin rule management endpoints

/admin/iam/* endpoints are guarded by _require_admin (the user's account-level role must be admin). These endpoints do not go through the standard IAM data gate — admin access to rule management is independent of data-access rules.


Integration

How mutation-engine uses IAM

The mutation engine (domain/mutation_engine.py) does not import from iam/ directly. IAM enforcement happens in server.py as a pre-flight check before any call into the mutation engine.

resolve_owner from the mutation engine is conceptually separate from IAM — it resolves the node-level ownership assignment used by the collaborative editing system, not the IAM access-rule system:

def resolve_owner(node_id: str, eff: dict, assignments: list) -> str | None:

It walks the ancestry chain in eff["parent_of"] and returns the username of the most-specific active ownership assignment, or None if the node is unassigned. This is checked by GET /me/can-edit and determines whether the user's lock on a node is exclusive.

validate_add_node enforces domain-level structural rules (schema, duplicate labels, parent ownership, cycle prevention):

def validate_add_node(cmd: dict, username: str, eff: dict, assignments: list) -> ValidationResult:

IAM write access is checked in server.py before validate_add_node is called. The mutation engine receives a request only if IAM has already passed.

How the API layer enforces permissions

The enforcement pattern in server.py is consistent:

  1. Authenticate: auth: CurrentUser (the canonical role in api/deps.py) extracts user_id.
  2. IAM gate: call _require_iam_write(auth.user_id, target_node_id) or inline has_write_access(...) before any mutation.
  3. Read filtering: call the relevant iam.* function and return only the filtered result.

The _THIS_DOMAIN server variable (set from EIDOS_DOMAIN env var) is always passed as the domain argument so all enforcement is scoped to the correct domain.


Error Responses

IAMValidationError — rule management (400)

Raised by validation functions in iam/admin.py when a rule payload is invalid. Caught in the /admin/iam/rules POST and PUT handlers and converted to a 400:

{
  "error": "invalid_request",
  "message": "Invalid role 'superuser'. Must be one of: ['admin', 'editor', 'viewer']"
}

Triggers: invalid role value, unknown domain, subtree_root not a known node ID, attempt to assign a rule to __super__.

Permission denied on write — 403

Raised when the user lacks write access on the target subtree. The body depends on which gate fired:

  • _iam_reject_if_locked (mutations router — node validate/preview/node-update, relation preview/commit):
{
  "detail": "Insufficient access"
}
  • _require_iam_write (Excel-import endpoints):
{
  "detail": "Insufficient write access on target"
}

HTTP status: 403 Forbidden in both cases.

Permission denied on read — 403

Raised inline when is_node_accessible returns False on a direct node fetch, or when _gate_proxy_domain blocks a cross-domain proxy request:

{
  "detail": "Access denied to domain 'secondary-domain'"
}

HTTP status: 403 Forbidden.

Filtered responses (no error)

When a user has partial access — for example, a viewer grant on one subtree of a larger tree — IAM does not return an error. The response is silently filtered:

  • The /tree endpoint returns only the accessible subtree plus restricted path ancestors.
  • The /nodes endpoint returns only accessible nodes.
  • The /relations endpoint returns only relations whose source is accessible; out-of-scope targets appear with restricted: true and no data fields.

The frontend receives a valid but narrowed response and should not present an error state to the user. The restricted flag on tree nodes and relation targets signals that content exists but is not visible to the current user.

Rule not found — 404

All /admin/iam/rules/{rule_id} endpoints return 404 when the rule ID does not exist or belongs to __super__:

{
  "detail": "Rule not found"
}