"""Per-client person visibility — the External API's population ceiling. The sibling of :mod:`app.api_clients.visibility` for people. A scope grant answers "*what kind* of people data may this client read" (``people.github:read``). It does NOT answer "*which* people". For a lookup-only surface that gap is harmless — the leave lookup makes the caller name its subjects, so knowing who to ask about is itself the boundary. The GitHub mapping endpoint breaks that property: it can be asked by department, i.e. it LISTS people. Without a ceiling one leaked token enumerates the whole company roster. ``api_clients.person_visibility`` (migration 513) holds a **person filter** in exactly the shape the identity-consumer pull surface already uses (:mod:`app.core.person_filters`): .. code-block:: json {"departments": [""], "include_subtree": true, "employment_statuses": ["active", "left"]} ``departments`` holds HRIS **org-node source ids** — the value ``IdentityOrgMembership.org_node_source_id`` carries and the shared ``PersonFilterEditor`` emits — NOT Helix row UUIDs. The filter engine compares them by exact string, so a wrong-kind identifier yields a ceiling that matches nobody; the governance layer existence-checks them on save rather than letting that ship silently. * ``NULL`` / ``{}`` = unrestricted — the pre-existing behaviour, so every client registered before this column shipped keeps working unchanged. * A restricted client can never see outside its filter, whatever it asks for: the request's own selectors (emails / handles / departments) are **intersected** with the ceiling, never unioned. A person outside it is reported exactly like a person who does not exist, so the surface never becomes an existence oracle for people the client may not read. Reusing ``FilterConfig`` rather than inventing a second vocabulary is deliberate: one engine evaluates both egress planes, one admin editor (``PersonFilterEditor``) edits both, and the push/pull/external surfaces cannot drift apart on what "population" means. This module stays SQL-free — it owns the wire shape (validate / normalize / describe). Resolving the shape against person rows is the router's job, through the shared filter engine. """ from __future__ import annotations import hashlib import uuid as uuid_mod from typing import Any from app.core.person_filters import FilterConfig, FilterError #: Dimensions a person-visibility payload may carry — the FULL #: ``FilterConfig`` vocabulary, deliberately not a subset: the admin edits #: this through the shared ``PersonFilterEditor``, which emits every #: dimension, so an allowlist narrower than the editor would reject a #: filter the UI itself produced. #: #: An unknown key is REJECTED rather than ignored: a typo'd #: ``{"department": [...]}`` would normalize to "no constraints" and #: silently hand the client the whole roster — precisely the failure this #: exists to prevent. (Same reasoning as ``CLOUD_VISIBILITY_KEYS``.) PERSON_VISIBILITY_KEYS: tuple[str, ...] = ( "include_offboarded", "departments", "include_subtree", "countries", "company_codes", "employment_statuses", "job_title_includes", "email_suffixes", "employee_type_codes", "employee_no_pattern", "person_ids", "mode", ) #: Dimensions that NARROW the population. ``include_offboarded`` / #: ``include_subtree`` / ``mode`` are absent on purpose — they widen or #: re-combine, they never constrain on their own, so a payload carrying #: only those is still "unrestricted". _NARROWING_KEYS: tuple[str, ...] = ( "departments", "countries", "company_codes", "employment_statuses", "job_title_includes", "email_suffixes", "employee_type_codes", "employee_no_pattern", "person_ids", ) def _canonical_uuid(value: Any) -> str: """Lower-cased, hyphenated UUID string; the raw value untouched when it isn't a UUID (the caller's existence check reports that clearly).""" try: return str(uuid_mod.UUID(str(value).strip())) except (ValueError, AttributeError, TypeError): return str(value).strip() def normalize_person_visibility(payload: Any) -> dict[str, Any] | None: """Validate + canonicalize a ``person_visibility`` payload. Returns the canonical dict, or ``None`` for "unrestricted". Raises ``ValueError`` with an actionable message on malformed input. Validation is delegated to :class:`FilterConfig` — the same parser the sync targets and directory consumers go through — so an admin cannot save a filter here that the evaluator would later reject or read differently. """ if payload is None: return None if not isinstance(payload, dict): raise ValueError( "person_visibility must be an object like " '{"departments": [""], "include_subtree": true}', ) unknown = sorted(set(payload) - set(PERSON_VISIBILITY_KEYS)) if unknown: raise ValueError( f"person_visibility: unknown key(s) {', '.join(unknown)}; " f"allowed keys are {', '.join(PERSON_VISIBILITY_KEYS)}", ) try: cfg = FilterConfig.from_dict(payload) except FilterError as exc: raise ValueError(f"person_visibility: {exc}") from exc # Drop empty dimensions so "unrestricted" has exactly one DB # representation, and so the audit diff doesn't read `{"countries": # []}` → `{}` as a change of grant. result: dict[str, Any] = {} if cfg.departments: result["departments"] = sorted(cfg.departments) # Only meaningful alongside a department list; recorded verbatim # because it flips the grant between one node and a whole subtree. result["include_subtree"] = bool(cfg.include_subtree) if cfg.countries: result["countries"] = sorted(cfg.countries) if cfg.company_codes: result["company_codes"] = sorted(cfg.company_codes) if cfg.employment_statuses: result["employment_statuses"] = sorted(cfg.employment_statuses) if cfg.job_title_includes: result["job_title_includes"] = sorted(cfg.job_title_includes) if cfg.email_suffixes: result["email_suffixes"] = sorted(cfg.email_suffixes) if cfg.employee_type_codes: result["employee_type_codes"] = sorted(cfg.employee_type_codes) if cfg.employee_no_pattern: result["employee_no_pattern"] = cfg.employee_no_pattern if cfg.person_ids: # Canonicalized to the DB's own UUID form. The filter engine # compares person ids as exact strings, so an uppercase or # unhyphenated spelling would store a ceiling that matches nobody # — and, worse, the governance existence-check would find the row # and still report "unknown person id", sending the admin hunting # a person who is right there. # De-duplicated AFTER canonicalizing: two spellings of one id # (uppercase, unhyphenated) are one person, and storing both would # inflate `person_count` in the audit row for a grant that names a # single individual. result["person_ids"] = sorted({_canonical_uuid(p) for p in cfg.person_ids}) if cfg.include_offboarded: result["include_offboarded"] = True # ``mode`` only changes meaning with 2+ active dimensions, but it is # stored whenever non-default so the saved filter round-trips exactly. if result and cfg.mode != "and": result["mode"] = cfg.mode return result or None def person_visibility_is_unrestricted(value: Any) -> bool: """True when ``value`` grants the whole roster. Tolerant of legacy / hand-edited rows (``{}``, ``{"countries": []}``) so callers need not re-normalize before asking. ``include_offboarded`` alone is NOT a narrowing — it widens — so it does not count as a constraint here. """ if not isinstance(value, dict): return True return not any(value.get(key) for key in _NARROWING_KEYS) #: Above this many named person ids the audit row records a count plus a #: truncated sample instead of the whole list — a 500-id grant would bury #: the change in noise. Departments are never truncated: a department #: swap is the highest-leverage widening there is (one team → a whole #: org) and the ids are few. _AUDIT_PERSON_ID_SAMPLE = 25 def person_visibility_audit_detail(value: Any) -> dict[str, Any]: """Summary for an audit row — legible, and lossless where it matters. The load-bearing requirement is that a **widening must never render identical to the grant it replaced**. An earlier version reduced ``departments`` to a bare count, so moving a client from one team to the whole engineering org wrote ``from`` and ``to`` as byte-identical dicts: the audit proved something changed and nothing about what. Same for an equal-length ``person_ids`` swap, and ``include_offboarded`` (which pulls departed staff into the ceiling) was dropped entirely. So: every narrowing dimension is recorded by VALUE, plus ``include_offboarded`` / ``include_subtree`` / ``mode``, which change what the same dimension values mean. Only ``person_ids`` is bounded, and it keeps its full count alongside the sample so a change of size is always visible. """ # Checked BEFORE the unrestricted early-return: a non-dict value is # "unrestricted" to that predicate, so clearing an unreadable ceiling # — the single largest widening on this surface, from *reads nobody* # to *reads the whole roster* — would otherwise audit as # ``from == to == unrestricted``, the exact fail-open lie this # function was hardened to stop telling. if value is not None and not isinstance(value, dict): return {"person_visibility": "misconfigured", "stored_type": type(value).__name__} if person_visibility_is_unrestricted(value): return {"person_visibility": "unrestricted"} # Both non-dict and unrestricted returned above, so this is a # populated filter; bound to a typed name so the projection below # reads as dict access rather than Any-shaped guesswork. assert isinstance(value, dict) # narrowed by the two returns above ceiling: dict[str, Any] = value detail: dict[str, Any] = {"person_visibility": "restricted"} if ceiling.get("departments"): detail["departments"] = list(ceiling["departments"]) detail["department_count"] = len(ceiling["departments"]) # Only meaningful next to a department list, but it flips the # grant between one node and an entire subtree. # Defaults to True, matching ``FilterConfig.include_subtree`` — a # hand-edited row that omits the key means "this node AND its # subtree" to the engine, and rendering it as "just this node" # would understate the grant in both the audit row and /me. detail["include_subtree"] = bool(ceiling.get("include_subtree", True)) for key in ( "countries", "company_codes", "employment_statuses", "job_title_includes", "email_suffixes", "employee_type_codes", ): if ceiling.get(key): detail[key] = list(ceiling[key]) if ceiling.get("employee_no_pattern"): detail["employee_no_pattern"] = ceiling["employee_no_pattern"] if ceiling.get("person_ids"): ids = list(ceiling["person_ids"]) detail["person_count"] = len(ids) detail["person_ids"] = ids[:_AUDIT_PERSON_ID_SAMPLE] if len(ids) > _AUDIT_PERSON_ID_SAMPLE: detail["person_ids_truncated"] = True # A digest over the WHOLE list, because the sample alone # re-opens the very hole this function exists to close: swap # one id past position 25 and the count, the sample and the # flag are all unchanged, so the audit diff reads as a no-op # again. The digest is not reversible — it identifies the # grant, it does not disclose its members. detail["person_ids_digest"] = hashlib.sha256( "\n".join(sorted(ids)).encode("utf-8"), ).hexdigest()[:16] # A widening that ONLY flips include_offboarded (departed staff enter # the ceiling) must not read as a no-op either. if ceiling.get("include_offboarded"): detail["include_offboarded"] = True if ceiling.get("mode") and ceiling["mode"] != "and": detail["mode"] = ceiling["mode"] return detail __all__ = [ "PERSON_VISIBILITY_KEYS", "normalize_person_visibility", "person_visibility_audit_detail", "person_visibility_is_unrestricted", ]