platform/core/perm_scope.py

"""core/perm_scope.py — the permission WALL for table modules (wave 15, C-PERM).

ONE place answers the three questions a restricted account raises on a grid surface:

    may_access(user, module)          may they open it at all?
    visible_fields(fields, u, mod)    which COLUMNS may they receive?
    apply_row_scope(rows, u, mod, …)  which ROWS may they receive?

plus one that exists only because of how the pool is built:

    derive_pool_scope(user, module)   which (team_id, agent) must the pool be BUILT with?

Both hosts call these — `aios-web/api` (`grid_assembly`) and `app.py` (`_table_grid`) — because
a wall that exists on one runtime and not the other is not a wall. `core/perms.py` stays what it
is (module GRANTS + the legacy BU derivation); this module is the row/field/pushdown layer that
sits on top, and it is deliberately a separate file so the legacy readers can keep their
semantics untouched while this one fails closed.

────────────────────────────────────────────────────────────────────────────────────────────
THE RECORD

    user['perms']   = {'<module_key>': {'access': bool,
                                        'filter': {'conj'?: 'and'|'or', 'nodes': [...]} | None,
                                        'hiddenFields': ['<field key>', ...]}}
    user['perms_v'] = 1        stamped by the migration and by every write

`perms_v` is the EXPLICIT-RESOLUTION marker, and it is here because of `permissioning.md`
Part II gap #5: "`allowed_modules() is None` is fail-open by default … make 'resolved:
unrestricted' an explicit value so absence/uncertainty DENIES." The same class already shipped
twice in this codebase (`modules: []` and `bus: [<unknown id>]` both read as UNRESTRICTED —
`routes_admin.py`'s own docstring documents both). So:

  * `perms_v` ABSENT  → the record is UN-MIGRATED, and the LEGACY wall applies unchanged
    (`core.perms` module grants + the `bus`/`agent` query scope). That is not fail-open: it is
    today's real wall, and it bounds the rollout window to "until the migration runs".
  * `perms_v == 1` and the module has NO entry → **DENY**. Absence now means what it says.
  * `role == 'admin'` bypasses all of it — which is also what keeps BREAK-GLASS alive.
    `deps._user_for` hands back a hardcoded master dict on a store outage
    (`{'username':'admin','role':'admin','bus':'all','modules':'all'}`) that will never carry a
    perms block; without this clause an explicit-marker scheme locks the owner out of their own
    product at exactly the moment the store is broken.
"""
import re

import core.perms as perms

#: `aios_grid._FORMULA_REF`'s pattern, restated rather than imported: this module is imported by
#: the API's request path and `aios_grid` pulls in the whole grid stack. Same regex, one line,
#: and `verify_api` asserts the two agree so it cannot drift into a different grammar.
_FORMULA_REF = re.compile(r"\{([^{}]*)\}")

PERMS_VERSION = 1


def _rec(user):
    return user if isinstance(user, dict) else {}


def is_migrated(user):
    """True once this record carries an explicit resolution. See the module docstring."""
    return int(_rec(user).get('perms_v') or 0) >= PERMS_VERSION


def entry(user, module):
    """This user's declared permissions for `module`, or None if nothing is declared.

    None is AMBIGUOUS on purpose and every caller must resolve it against `is_migrated`:
    on a migrated record it means DENY, on a legacy one it means "ask the old wall".
    """
    p = _rec(user).get('perms')
    if not isinstance(p, dict):
        return None
    e = p.get(module)
    return e if isinstance(e, dict) else None


def may_access(user, module):
    """May this account open `module` at all? Fail-closed on a migrated record."""
    if perms.is_admin(user):
        return True
    e = entry(user, module)
    if e is not None:
        return bool(e.get('access', True))
    if is_migrated(user):
        # Migrated and undeclared = denied. This is the whole point of the marker.
        return False
    return perms.may_open(user, module)          # legacy record: the old grant wall


# ── FIELDS ───────────────────────────────────────────────────────────────────────────────────
def hidden_keys(user, module, fields):
    """The TRANSITIVE closure of hidden field keys (C-PERM amendment 5).

    ⛔ WHY A CLOSURE AND NOT A SET DIFFERENCE. A formula field is computed in the BROWSER
    (`formulaEngine.ts`, injected by `computedRows`) from `{ref}`s into other columns, and a
    measure column's value arrives precomputed in `derived`. So hiding field X has exactly three
    possible outcomes and only one of them is coherent:

        strip X, keep formulas    → every formula over X computes blank or wrong, silently
        keep X's value for them   → X has leaked, wearing a formula's name
        strip X AND its dependents→ the only honest answer

    So a hidden field drags every formula that references it — and every formula that references
    THAT formula, hence the fixpoint loop — out of the payload with it.

    ⚠ This runs on every assembly, so it is a fixpoint over a handful of custom fields, not a
    graph library. `MAX_PASSES` bounds a reference cycle the client would refuse to evaluate
    anyway; without it a self-referential pair would spin here.
    """
    e = entry(user, module)
    if perms.is_admin(user) or not e:
        return frozenset()
    hidden = {str(k) for k in (e.get('hiddenFields') or ()) if k}
    if not hidden:
        return frozenset()

    refs = {}
    for f in fields or ():
        if not isinstance(f, dict) or not f.get('key'):
            continue
        expr = f.get('formula')
        if isinstance(expr, str) and expr:
            refs[f['key']] = {m.strip() for m in _FORMULA_REF.findall(expr) if m.strip()}

    MAX_PASSES = 12
    for _ in range(MAX_PASSES):
        grew = False
        for key, deps in refs.items():
            if key not in hidden and deps & hidden:
                hidden.add(key)
                grew = True
        if not grew:
            break
    return frozenset(hidden)


def visible_fields(fields, user, module):
    """`fields` minus the hidden closure. Order preserved — the column order is the user's."""
    hide = hidden_keys(user, module, fields)
    if not hide:
        return list(fields or ())
    return [f for f in (fields or ())
            if not (isinstance(f, dict) and f.get('key') in hide)]


def strip_row(row, hide):
    """Drop hidden keys from ONE assembled row. Cheap enough to run per row, and it must run
    per row: the field list and the row payload are two different wires, and stripping only the
    first would leave the value sitting in the second where anyone can read it."""
    if not hide or not isinstance(row, dict):
        return row
    return {k: v for k, v in row.items() if k not in hide}


# ── ROWS ─────────────────────────────────────────────────────────────────────────────────────
def apply_row_scope(rows, user, module, fields, ctx=None):
    """The rows this account may receive: `filter_eval.permits` over the permanent filter.

    `permits`, never `matches` — an unanswerable permanent filter DENIES rather than being
    ignored. See `harness/filter_eval`'s docstring for the field-rename walkthrough that makes
    the difference a leak rather than a preference.
    """
    if perms.is_admin(user):
        return list(rows or ())
    e = entry(user, module)
    tree = (e or {}).get('filter')
    if not tree:
        return list(rows or ())
    from harness import filter_eval as fe
    return [r for r in (rows or ()) if fe.permits(tree, r, fields, ctx)]


# ── THE PUSHDOWN ─────────────────────────────────────────────────────────────────────────────
#: `dba` value sets that pin a BU, and the Odoo team they pin. `_dba_attrs` emits exactly
#: {'Fisch','Royal','Both'} (blank for "no brand attributable in 24 months"), so a permission
#: filter admitting Fisch admits {Fisch, Both} — a Both customer IS a Fisch customer.
_DBA_TEAM = ((frozenset({'fisch', 'both'}), 5), (frozenset({'royal', 'both'}), 6))

#: Only equality pushes down. `FILTER_OPS` is single-valued (there is no column-level "is any
#: of" — that vocabulary belongs to cohort leaves and is disjoint), so a multi-value BU
#: condition arrives as an OR GROUP of `eq` leaves, handled by `_group_dba_team` below.
_PUSHDOWN_OPS = frozenset({'eq'})

#: ⭐ THE MODULES WHOSE FIELD VOCABULARY CAN EXPRESS A BUSINESS UNIT — i.e. whose canonical
#: contract carries a `dba` column. R1's "BU access is just a permanent filter" holds only on
#: these; on every other topic there is no column to write the condition against, so a filter
#: literally cannot say it and the record's `bus` remains the only place the fact lives.
#:
#: `product_data` is the counter-example that made this constant necessary: 19 fields, no brand
#: column, and a product's BU is a property of WHOSE ORDERS built its revenue rather than of the
#: SKU. Without the fallback below, a Fisch-only account read the product catalogue with
#: Fisch+Royal money on every row — the amendment-3 defect, arriving through the door amendment 3
#: was written to close.
#:
#: ⚠ A LIST THAT MIRRORS A JSON CONTRACT DRIFTS UNLESS SOMETHING CHECKS IT. `verify_api` asserts
#: membership here matches "this topic's contract has a `dba` field" for every governed module, so
#: a topic that grows or loses a brand column cannot silently keep the wrong rule. Named in this
#: module rather than derived from `aios_grid` on purpose: `perm_scope` is on the API's request
#: path and importing the grid stack to answer a two-element question is a cost per request.
BU_FILTERABLE_MODULES = frozenset({'customer_data'})


def derive_pool_scope(user, module):
    """`(team_id, agent)` the POOL must be BUILT with, derived from the permanent filter.

    ⛔ THIS EXISTS BECAUSE `team_id` SHAPES VALUES, NOT ROW MEMBERSHIP (C-PERM amendment 3).
    `modules/customer_data._pool_build` passes `team_id` into `cust._cust_rev` three times (YTD,
    LY, LTM) and into `cust._cadence_bulk`, so it decides what `rev`, `ly`, `ltm`, `aov`,
    `est_missed` and the derived `status` MEAN. Enforce a BU purely as a post-filter and a
    Fisch-only user keeps a correct-looking row LIST while every number on it silently becomes
    Fisch+Royal — worst for `dba = Both` customers, who are exactly the ones a BU filter admits.
    A pid-level reconciliation cannot see that; the values one can, and does.

    So the query-level pushdown SURVIVES — but as a DERIVATION OF the permanent filter rather
    than a second wall beside it, which is what keeps R1's "BU access is just a filter" true at
    the level the owner asked for it (one declaration, one UI, one engine).

    Pure, and recomputed per request rather than stored: a stored derivation drifts from the
    filter it came from, and then two things disagree about what an account may see.

    Reads TOP-LEVEL AND-conjunction leaves ONLY. A leaf under `or` guarantees no narrowing —
    `dba is Fisch OR revenue > 10` must not pin the pool to Fisch — so it never pushes down.
    Anything not recognised here simply is not pushed down; `apply_row_scope` still applies the
    whole tree, so the wall is unchanged either way. Belt AND braces, deliberately: the pushdown
    is what makes the VALUES right, `permits()` is what makes the ROWS right.

    ⭐ THE `bus` FALLBACK, AND WHY IT IS NOT A HOLE IN AMENDMENT 4. On a topic outside
    `BU_FILTERABLE_MODULES` there is no column a BU condition could be written against, so
    "the filter pins no team" cannot mean "the admin chose consolidated" — it is the only answer
    the filter language has. Resolving that silence as None returns the WIDER scope, which makes
    the current code fail-OPEN on the values axis for exactly the topic that cannot argue back.
    So the record's own `bus` answers instead, and the direction is what makes it safe: this can
    only ever REPLACE None (both units) with a pinned single unit. It never widens, it never
    touches `may_access`, and a `bus:'all'` account is unaffected because `scope_team_id` returns
    None for it — which is every account in tenant #0's registry except the one this shipped for.
    """
    if perms.is_admin(user):
        return None, None
    team_id, agent = _derive_from_filter(user, module)
    if team_id is None and module not in BU_FILTERABLE_MODULES:
        team_id = perms.scope_team_id(user)
    return team_id, agent


def _derive_from_filter(user, module):
    """`(team_id, agent)` the PERMANENT FILTER pins, before any fallback. Split out so the
    fallback has exactly one place to apply — the three exits below all mean "the filter pinned
    nothing", and a rule written at each of them is a rule that will one day be written at two."""
    e = entry(user, module)
    tree = (e or {}).get('filter')
    if not tree:
        # Un-migrated records still answer through the legacy derivation, so the old wall keeps
        # working until the migration has run.
        if not is_migrated(user):
            return perms.scope_team_id(user), perms.scope_agent(user)
        return None, None

    from harness import filter_eval as fe
    nodes, conj = fe.tree_parts(tree)
    if conj == 'or':
        return None, None

    team_id, agent = None, None
    for n in nodes:
        if not isinstance(n, dict):
            continue
        if isinstance(n.get('children'), list):
            # A top-level OR GROUP under an AND root IS a guaranteed narrowing — every row must
            # satisfy it — so it may push down, unlike a leaf under an OR ROOT (refused above).
            # This is the shape a multi-value BU condition actually takes; see `_group_dba_team`.
            tid = _group_dba_team(n)
            if tid is not None:
                team_id = tid if team_id in (None, tid) else None
            continue
        if n.get('op') not in _PUSHDOWN_OPS:
            continue
        col = n.get('colId')
        raw = n.get('value')
        if col == 'dba':
            vals = {v.strip().lower() for v in str(raw or '').split(',') if v.strip()}
            if not vals:
                continue
            for allowed, tid in _DBA_TEAM:
                if vals <= allowed:
                    # Both BUs named = no narrowing to push; leave it to the post-filter.
                    team_id = tid if team_id in (None, tid) else None
                    break
        elif col == 'agent':
            v = str(raw or '').strip()
            # A SET of agents cannot become the pool's single `agent_name`; the post-filter
            # handles it. Only an unambiguous single value pushes down.
            if v and ',' not in v:
                agent = v
    return team_id, agent


def _group_dba_team(group):
    """The team a top-level `or` group pins, or None.

    Recognises ONLY the exact shape "every child is a `dba eq <brand>` leaf" — the group the
    condition builder emits for a multi-value BU condition, and the one `perm_migrate` writes.
    Every OTHER group returns None and is left entirely to the post-filter: a group mixing `dba`
    with another column, or containing a nested group, does not pin a BU on its own, and
    guessing that it does would build the pool from the wrong book. Narrow by construction —
    the pushdown may only ever be an OPTIMISATION of a constraint the filter already expresses.
    """
    if group.get('conj') != 'or':
        return None
    children = group.get('children') or []
    if not children:
        return None
    vals = set()
    for c in children:
        if (not isinstance(c, dict) or isinstance(c.get('children'), list)
                or c.get('colId') != 'dba' or c.get('op') != 'eq'):
            return None
        v = str(c.get('value') or '').strip().lower()
        if not v:
            return None
        vals.add(v)
    for allowed, tid in _DBA_TEAM:
        if vals <= allowed:
            return tid
    return None