"""Per-client cloud visibility — the External API's resource whitelist. A scope grant answers "*what kind* of data may this client read" (``cloud.subscriptions:read``). It does NOT answer "*which* cloud accounts". Until this module existed, every holder of the cloud scope read the entire company inventory; once billing joins the surface that would mean one leaked token exposes the whole cloud spend picture. ``api_clients.cloud_visibility`` narrows a client to a set of cloud subscriptions: .. code-block:: json {"providers": ["aws", "gcp"], "subscription_ids": ["", "…"]} * The two lists are a **union** — "every subscription of these providers, plus these individually named ones". * ``NULL`` / ``{}`` = unrestricted. That is the pre-existing behaviour, so every client registered before this column shipped keeps working unchanged. * A restricted client that matches nothing sees nothing. The resolver never degrades an unmatched whitelist into "unrestricted" — see :mod:`app.domains.cmdb.open_visibility`. * A stored value that is present but NOT a JSON object (a hand-edited row, a restore from an older shape) is **malformed**, and malformed fails CLOSED — the client reads nothing. It must never fall back to the ``NULL`` meaning of "unrestricted", which would hand a deliberately narrowed client the entire inventory. See :func:`cloud_visibility_is_malformed`. This module is deliberately SQL-free and domain-free: it owns the wire shape (parse / validate / normalize / describe) so the governance router, the auth context, and the cmdb-side resolver all agree on one canonical form. Resolving the shape against actual cloud rows is cmdb's job and lives in that domain. """ from __future__ import annotations import re import uuid as uuid_mod from typing import Any #: Keys the object accepts. An unknown key is rejected rather than #: ignored: a typo'd ``{"provider": ["aws"]}`` would normalize to "no #: constraints" and silently hand the client the whole inventory — #: exactly the failure this feature exists to prevent. CLOUD_VISIBILITY_KEYS = ("providers", "subscription_ids") #: Bounds keep one client row (and the IN-clause it becomes) sane. CLOUD_VISIBILITY_MAX_PROVIDERS = 32 CLOUD_VISIBILITY_MAX_SUBSCRIPTIONS = 500 _PROVIDER_RE = re.compile(r"^[a-z0-9][a-z0-9_-]{0,62}$") def normalize_cloud_visibility(payload: Any) -> dict[str, list[str]] | None: """Validate + canonicalize a ``cloud_visibility`` payload. Returns the canonical dict (lower-cased, de-duplicated, sorted, with empty lists dropped) or ``None`` for "unrestricted". Raises ``ValueError`` with an actionable message on any malformed input. Canonical form matters beyond tidiness: the audit trail diffs the stored value, so ``["aws","gcp"]`` and ``["gcp","aws"]`` must not read as a change of grant. """ if payload is None: return None if not isinstance(payload, dict): raise ValueError( "cloud_visibility must be an object like " '{"providers": ["aws"], "subscription_ids": [""]}', ) unknown = sorted(set(payload) - set(CLOUD_VISIBILITY_KEYS)) if unknown: raise ValueError( f"cloud_visibility: unknown key(s) {', '.join(unknown)}; " f"allowed keys are {', '.join(CLOUD_VISIBILITY_KEYS)}", ) providers = _normalize_providers(payload.get("providers")) subscription_ids = _normalize_subscription_ids(payload.get("subscription_ids")) result: dict[str, list[str]] = {} if providers: result["providers"] = providers if subscription_ids: result["subscription_ids"] = subscription_ids # ``{}`` / ``{"providers": []}`` collapse to NULL so "unrestricted" # has exactly one representation in the DB. return result or None def _normalize_providers(raw: Any) -> list[str]: if raw is None: return [] if not isinstance(raw, list): raise ValueError("cloud_visibility.providers must be a list of strings") out: set[str] = set() for item in raw: if not isinstance(item, str): raise ValueError("cloud_visibility.providers must be a list of strings") value = item.strip().lower() if not value: continue if not _PROVIDER_RE.match(value): raise ValueError( f"cloud_visibility.providers: {item!r} is not a valid provider " "key (lowercase letters/digits/dash/underscore)", ) out.add(value) if len(out) > CLOUD_VISIBILITY_MAX_PROVIDERS: raise ValueError( "cloud_visibility.providers: at most " f"{CLOUD_VISIBILITY_MAX_PROVIDERS} providers", ) return sorted(out) def _normalize_subscription_ids(raw: Any) -> list[str]: if raw is None: return [] if not isinstance(raw, list): raise ValueError( "cloud_visibility.subscription_ids must be a list of subscription " "UUIDs (the Helix id, not the cloud-native account number)", ) out: set[str] = set() for item in raw: if not isinstance(item, str) or not item.strip(): raise ValueError( "cloud_visibility.subscription_ids must be a list of " "subscription UUIDs", ) try: parsed = uuid_mod.UUID(item.strip()) except (ValueError, AttributeError, TypeError) as exc: raise ValueError( f"cloud_visibility.subscription_ids: {item!r} is not a valid " "Helix subscription id (UUID). Cloud-native account numbers " "are not accepted here — they are not unique across tenants.", ) from exc out.add(str(parsed)) if len(out) > CLOUD_VISIBILITY_MAX_SUBSCRIPTIONS: raise ValueError( "cloud_visibility.subscription_ids: at most " f"{CLOUD_VISIBILITY_MAX_SUBSCRIPTIONS} subscriptions; use the " "providers list for a whole-tenant grant", ) return sorted(out) def cloud_visibility_is_unrestricted(value: Any) -> bool: """True when ``value`` grants the whole cloud inventory. Tolerant of legacy / hand-edited rows (``{}``, ``{"providers": []}``) so callers need not re-normalize before asking. """ if not isinstance(value, dict): return True return not any(value.get(key) for key in CLOUD_VISIBILITY_KEYS) def cloud_visibility_is_malformed(value: Any) -> bool: """True when a stored whitelist is present but unreadable as one. ``NULL`` is a legitimate grant ("unrestricted"); anything that is not ``NULL`` and not a JSON object is a value no caller can evaluate — a hand-edited row, a restore from an older shape, a future writer that stored the object serialized. Such a value must fail **closed** (the client sees nothing), never degrade to the ``NULL`` meaning, which would silently widen a narrowed client to the whole inventory. Sole definition of the predicate: the auth context, the cmdb resolver, the audit summary and the governance response all ask this function, so "malformed" cannot come to mean different things on the data plane and the console. """ return value is not None and not isinstance(value, dict) def cloud_visibility_audit_detail(value: Any) -> dict[str, Any]: """Compact, greppable summary for an audit row. Records the shape of the grant (which providers, how many named subscriptions) rather than 500 UUIDs — the full value lives on the client row; the audit trail needs the change to be legible. """ # Checked BEFORE the unrestricted branch: a non-dict value reads as # "unrestricted" to that predicate, so clearing an unreadable # whitelist — the single largest widening on this surface, from # *reads nothing* to *reads the whole cloud inventory* — would # otherwise audit as ``from == to == unrestricted``: proof that # something changed and nothing about what. if cloud_visibility_is_malformed(value): return { "cloud_visibility": "misconfigured", "stored_type": type(value).__name__, } if cloud_visibility_is_unrestricted(value): return {"cloud_visibility": "unrestricted"} # Malformed and unrestricted both returned above, so this is a # populated whitelist. assert isinstance(value, dict) # narrowed by the two returns above providers = list(value.get("providers") or []) subs = list(value.get("subscription_ids") or []) return { "cloud_visibility": "restricted", "providers": providers, "subscription_count": len(subs), } __all__ = [ "CLOUD_VISIBILITY_KEYS", "CLOUD_VISIBILITY_MAX_PROVIDERS", "CLOUD_VISIBILITY_MAX_SUBSCRIPTIONS", "cloud_visibility_audit_detail", "cloud_visibility_is_malformed", "cloud_visibility_is_unrestricted", "normalize_cloud_visibility", ]