"""Client for a Rego (OPA) permissions service.
A codegen-opa permissions service is a stateless Open Policy Agent
deployment: it carries the policy (the generic RBAC engine plus
any custom rules) but no data. The roles and role bindings live
in the backend's own database; this module ships them -- together
with the subject, the action, and the resource -- to OPA on every
call.
Three query modes:
* **Point check** -- a decision about one resource (get / create /
update / delete). :meth:`OpaClient.check` POSTs the decision
``input`` to ``/v1/data/<opa_package>/decision`` and returns a
:class:`Decision`.
* **Bulk check** -- many ``(action, resource)`` decisions for one
subject in a single round-trip (the actions-envelope case:
every row of a list page x every action).
:meth:`OpaClient.check_many` POSTs to
``/v1/data/<opa_package>/decisions`` and returns one
:class:`Decision` per item.
* **List filter** -- a collection request where every candidate
row must be checked. :meth:`OpaClient.compile_filter` asks
OPA to *partially evaluate* the ``allow`` rule with the resource
left unknown (the ``/v1/compile`` API). OPA returns residual
conditions on the resource; :meth:`FilterResult.to_sqlalchemy`
turns them into a SQLAlchemy ``WHERE`` clause the backend folds
into the list query. One round-trip filters the whole
collection -- no per-row check.
Requires the ``opa`` extra (``pip install 'fsh-lib[opa]'``)
for :mod:`httpx`.
Example -- point check::
client = OpaClient("http://opa:8181", opa_package="authz")
decision = await client.check(
subject=Subject("user", "alice"),
action="task:update",
resource=ResourceRef("Task", "t-1", {"created_by": "alice"}),
roles={"editor": {"permissions": ["task:read", "task:update"]}},
bindings=[RoleBinding(Subject("user", "alice"), "editor")],
)
if not decision.permit:
raise HTTPException(status_code=403)
Example -- list filter::
result = await client.compile_filter(
subject=Subject("user", "alice"),
action="task:list",
resource_type="Task",
roles=roles,
bindings=bindings,
)
stmt = select(Task).where(result.to_sqlalchemy({"id": Task.id}))
"""
from __future__ import annotations
from dataclasses import dataclass, field
from typing import TYPE_CHECKING, Any, Self
import httpx
from sqlalchemy import and_, false, not_, or_, true
from fsh_lib.rbac import (
RoleBindingScopeMixin, # noqa: F401 (re-export; was defined here)
)
if TYPE_CHECKING:
from collections.abc import Mapping, Sequence
from sqlalchemy.sql.elements import ColumnElement
#: Default per-request timeout, in seconds. A permissions check
#: sits on the request path, so the timeout is short -- a slow
#: service should fail fast and let the caller apply its
#: fail-open / fail-closed policy.
_DEFAULT_TIMEOUT = 5.0
[docs]
class OpaError(RuntimeError):
"""A permissions-service call failed.
Raised when the service is unreachable, times out, answers
with a non-2xx status, or returns a residual the translator
cannot turn into a SQL filter. The caller decides whether an
error means "deny" (fail-closed, the safe default for authz)
or "allow" (fail-open) -- this module never silently picks
one.
"""
[docs]
@dataclass(frozen=True)
class Subject:
"""The principal a decision is made for.
In the token-session model the decision subject is the *token*
(``type="token"``, ``id`` = the token id), and the user it was
minted for rides along under :attr:`user`. RBAC matching is on
the ``(type, id)`` pair -- i.e. the token's own snapshot of role
bindings; the nested ``user`` is identity / audit / ownership
context only and never participates in role matching.
Attributes:
type: ``"token"`` (the default principal in the token-session
model) or ``"user"``. A binding for one kind never
satisfies a request for the other -- the RBAC engine
matches on the pair.
id: Stable identifier of the principal -- a token id in the
token-session model, or a user id for a bare ``"user"``
subject.
user: Optional nested user identity folded into
``input.subject.user`` so Rego rules (and ownership
checks) can read the human behind a token without that
identity affecting role matching. ``None`` omits the
key entirely.
"""
type: str
id: str
user: Mapping[str, Any] | None = None
[docs]
@dataclass(frozen=True)
class ResourceRef:
"""The resource a decision concerns.
Attributes:
type: Resource type, e.g. ``"Task"``. Matched against the
permission catalogue and against object-scoped
bindings.
id: Resource instance id, or ``None`` for a
collection-level decision (a ``list`` / ``create``
where no single instance exists yet).
attributes: Extra fields folded into ``input.resource`` so
custom Rego rules can read them -- e.g.
``{"created_by": "alice"}`` for an ownership rule.
``type`` and ``id`` always take precedence over a
same-named attribute key.
"""
type: str
id: str | None = None
attributes: Mapping[str, Any] = field(default_factory=dict)
[docs]
class ScopeError(ValueError):
"""A role binding's attribute scope failed strict validation."""
[docs]
def scope_fields(opa_rep_class: type) -> dict[str, type]:
"""Allow-list of scopable fields for a resource's OPA representation.
Derived from the generated opa-rep model (which is itself generated
from the resource spec's ``opa`` representation): every field
except the ``type`` discriminator, mapped to the Python type its
value has at the OPA/JSON boundary (``uuid`` / ``datetime`` -> str,
``int`` -> int, ...). Pass to :meth:`RoleBinding.from_row` as
``allowed`` so a binding's scope is validated against the very
projection the policy filters on -- one source of truth (the
spec's ``opa`` rep), never a hand-listed dict.
``opa_rep_class`` only needs a Pydantic-style ``model_json_schema``
classmethod (every generated opa-rep model has one).
"""
json_py: dict[str, type] = {
"string": str,
"integer": int,
"number": float,
"boolean": bool,
}
fields: dict[str, type] = {}
properties = opa_rep_class.model_json_schema().get("properties", {})
for name, prop in properties.items():
if name == "type": # the resource-type discriminator, not a scope field
continue
json_type = _json_boundary_type(prop)
if json_type in json_py:
fields[name] = json_py[json_type]
return fields
def _json_boundary_type(prop: Mapping[str, Any]) -> str | None:
"""JSON-schema type of a property, unwrapping a nullable ``anyOf``."""
if "type" in prop:
return prop["type"]
for sub in prop.get("anyOf", []):
sub_type = sub.get("type")
if sub_type not in (None, "null"):
return sub_type
return None
[docs]
def parse_scope_attrs(
attrs: Mapping[str, Any],
allowed: Mapping[str, type],
) -> dict[str, Any]:
"""Strictly validate a binding's ``object.attrs``.
Checked against the resource's scopable fields and returned
unchanged so it can be piped straight into :class:`ResourceRef`.
Fail-closed, no coercion: every key must be a declared scopable
field and every value must be an *exact* type match (``type(v) is
T`` -- so a ``bool`` is never accepted for an ``int`` column, and
a stringified number is never coerced). An empty scope is
rejected too: an attribute binding with no predicate would widen
to the whole type.
Raises:
ScopeError: empty scope, an unknown key, or a value whose
type is not a direct match.
"""
if not attrs:
msg = "attribute scope is empty; use a type-scoped binding instead"
raise ScopeError(msg)
validated: dict[str, Any] = {}
for key, value in attrs.items():
if key not in allowed:
msg = f"unknown scope field {key!r}; allowed: {sorted(allowed)}"
raise ScopeError(msg)
expected = allowed[key]
if expected is not object and type(value) is not expected:
msg = (
f"scope field {key!r} expects {expected.__name__}, "
f"got {type(value).__name__} ({value!r})"
)
raise ScopeError(msg)
validated[key] = value
return validated
[docs]
@dataclass(frozen=True)
class RoleBinding:
"""A role bound to a subject, optionally scoped to a resource.
Attributes:
subject: The principal the role is granted to.
role: Name of the granted role. Must be a key of the
``roles`` catalogue passed to the query method.
scope: ``None`` for a global binding (applies to every
resource); a :class:`ResourceRef` with no ``id`` for a
type-scoped binding (every instance of a type); a
:class:`ResourceRef` with an ``id`` for an
instance-scoped binding; a :class:`ResourceRef` whose
``attributes`` are set for an attribute-scoped binding
(every instance whose field equals the bound value,
e.g. ``ResourceRef("Category",
attributes={"department_id": "D"})`` -> one department).
Attribute scope rides as ``object.attrs`` and only grants
through a resource rule that completes the predicate
(``rbac.scoped_grant``); list endpoints must add the
field to the compile ``unknowns``.
"""
subject: Subject
role: str
scope: ResourceRef | None = None
[docs]
@classmethod
def from_row(
cls,
subject: Subject,
role: str,
*,
object_type: str | None = None,
object_id: str | None = None,
scope_attrs: Mapping[str, Any] | None = None,
allowed: Mapping[str, type] | None = None,
) -> Self:
"""Assemble a validated binding from binding-row columns.
The ``load_bindings`` companion to
:class:`RoleBindingScopeMixin`.
* ``object_type is None`` -> global binding (every resource).
* ``object_type`` set, no ``scope_attrs`` -> type- or
instance-scoped (per ``object_id``).
* ``scope_attrs`` set -> attribute-scoped; validated against
``allowed`` (the resource's scopable fields, e.g. from
:func:`scope_fields`) when given, via
:func:`parse_scope_attrs`.
``None`` values in ``scope_attrs`` are dropped first -- a
``None`` dimension means "all" on that axis, so a binding that
scopes no dimension collapses to a plain type scope rather
than raising the empty-attrs error.
"""
if object_type is None:
return cls(subject, role)
attrs = {k: v for k, v in (scope_attrs or {}).items() if v is not None}
if attrs and allowed is not None:
attrs = parse_scope_attrs(attrs, allowed)
return cls(
subject, role, ResourceRef(object_type, object_id, attributes=attrs)
)
[docs]
@dataclass(frozen=True)
class Decision:
"""The answer a point permissions check returns.
Attributes:
permit: The bottom line. ``True`` iff some grant applies
and no veto overrides it -- the value the backend
gates on.
allow: ``True`` iff a grant applied (generic RBAC or a
custom allow rule), before vetoes.
deny: ``True`` iff a custom deny rule vetoed the request.
raw: The full ``result`` object the service returned, for
logging or richer custom decisions.
"""
permit: bool
allow: bool
deny: bool
raw: Mapping[str, Any]
[docs]
@classmethod
def from_result(cls, result: Mapping[str, Any]) -> Decision:
"""Build a :class:`Decision` from a decision ``result`` object.
Missing keys default to the safe value (``False``): a
service that answers with a partial document is treated as
a denial rather than a silent allow.
"""
return cls(
permit=bool(result.get("permit", False)),
allow=bool(result.get("allow", False)),
deny=bool(result.get("deny", False)),
raw=result,
)
[docs]
@dataclass(frozen=True)
class Condition:
"""One residual constraint on a resource field.
A :class:`Condition` is the translated form of a single
comparison OPA left unresolved when it partially evaluated the
policy -- e.g. ``input.resource.id == "t-1"`` becomes
``Condition("id", "eq", "t-1")``.
Attributes:
field: The resource field the constraint is on -- the path
after ``input.resource.``, dotted for a nested field.
op: The comparison: ``"eq"``, ``"ne"``, ``"lt"``,
``"le"``, ``"gt"``, ``"ge"``, or ``"in"``.
value: The literal the field is compared against (a list
for ``"in"``).
negated: ``True`` when OPA emitted the expression negated.
"""
field: str
op: str
value: Any
negated: bool = False
#: Builds a SQLAlchemy clause from a :class:`Condition`'s op.
_CLAUSE_BUILDERS = {
"eq": lambda col, value: col == value,
"ne": lambda col, value: col != value,
"lt": lambda col, value: col < value,
"le": lambda col, value: col <= value,
"gt": lambda col, value: col > value,
"ge": lambda col, value: col >= value,
"in": lambda col, value: col.in_(value),
}
[docs]
@dataclass(frozen=True)
class FilterResult:
"""A compiled list policy: which rows a subject may see.
The outcome of partially evaluating the ``allow`` rule with
the resource unknown. It is one of three shapes:
* **always allow** -- no constraint; every row passes.
* **always deny** -- the policy can never hold; no row passes.
* **conditional** -- :attr:`conjunctions` is an OR of ANDs of
:class:`Condition` objects (disjunctive normal form, the
shape OPA's ``/v1/compile`` returns).
Attributes:
always_allow: Every row passes; :meth:`to_sqlalchemy`
returns a tautology.
always_deny: No row passes; :meth:`to_sqlalchemy` returns
a contradiction.
conjunctions: OR-of-ANDs residual. Each inner tuple is a
conjunction of :class:`Condition` objects; a row
passes when it satisfies *any* conjunction.
"""
always_allow: bool
always_deny: bool
conjunctions: tuple[tuple[Condition, ...], ...]
[docs]
def to_sqlalchemy(
self,
columns: Mapping[str, ColumnElement[Any]],
) -> ColumnElement[bool]:
"""Render the residual as a SQLAlchemy boolean clause.
Args:
columns: Maps a resource field name (as it appears in
:attr:`Condition.field`) to the SQLAlchemy column
it constrains, e.g. ``{"id": Task.id}``.
Returns:
A clause for a ``WHERE`` / ``.where()``: a tautology
when :attr:`always_allow`, a contradiction when
:attr:`always_deny`, otherwise the OR-of-ANDs.
Raises:
OpaError: A residual constrains a field absent from
*columns*.
"""
if self.always_deny:
return false()
if self.always_allow:
return true()
disjuncts = [
and_(*(self._clause(cond, columns) for cond in conjunction))
for conjunction in self.conjunctions
]
return or_(*disjuncts)
@staticmethod
def _clause(
cond: Condition,
columns: Mapping[str, ColumnElement[Any]],
) -> ColumnElement[bool]:
"""Render one :class:`Condition` as a SQLAlchemy clause."""
column = columns.get(cond.field)
if column is None:
msg = (
f"residual constrains resource field {cond.field!r}, "
f"which is not in the supplied column map "
f"{sorted(columns)}"
)
raise OpaError(msg)
clause = _CLAUSE_BUILDERS[cond.op](column, cond.value)
return not_(clause) if cond.negated else clause
[docs]
class OpaClient:
"""Async client for a codegen-opa Rego permissions service.
Wraps an :class:`httpx.AsyncClient` pointed at an OPA decision
server. Construct one per application (it is cheap to keep
open) and close it on shutdown via :meth:`aclose`, or use it
as an async context manager.
Args:
base_url: Base URL of the OPA server, e.g.
``"http://opa:8181"``.
opa_package: Rego package of the decision entrypoint. The
point-check path is ``/v1/data/<opa_package>/decision``;
the list filter compiles ``data.<opa_package>.allow``.
Defaults to ``"authz"``.
client: An existing :class:`httpx.AsyncClient` to use.
When ``None`` (the default) the :class:`OpaClient`
creates and owns one, and :meth:`aclose` closes it.
timeout: Per-request timeout in seconds. Ignored when an
explicit *client* is supplied.
"""
def __init__(
self,
base_url: str,
*,
opa_package: str = "authz",
client: httpx.AsyncClient | None = None,
timeout: float = _DEFAULT_TIMEOUT,
) -> None:
"""Store the decision paths and the (owned or borrowed) client."""
self._opa_package = opa_package
self._decision_path = f"/v1/data/{opa_package}/decision"
self._bulk_path = f"/v1/data/{opa_package}/decisions"
self._owns_client = client is None
self._client = client or httpx.AsyncClient(
base_url=base_url,
timeout=timeout,
)
async def __aenter__(self) -> Self:
"""Enter the async context, returning ``self``."""
return self
async def __aexit__(self, *_exc: object) -> None:
"""Close an owned client on context exit."""
await self.aclose()
[docs]
async def aclose(self) -> None:
"""Close the underlying client iff this instance owns it."""
if self._owns_client:
await self._client.aclose()
[docs]
async def check(
self,
*,
subject: Subject,
action: str,
resource: ResourceRef,
roles: Mapping[str, Any],
bindings: Sequence[RoleBinding],
format: str | None = None, # noqa: A002
representation: str | None = None,
) -> Decision:
"""Ask the service whether *subject* may take *action*.
A point check -- one decision about one resource.
Args:
subject: The principal making the request.
action: The permission string being checked, e.g.
``"task:update"``.
resource: The resource the action concerns.
roles: The role catalogue -- ``{role_name: {"permissions":
[...]}}``. The backend loads this from its own
store; only the roles relevant to *bindings* need
be present.
bindings: The subject's role bindings, loaded by the
backend from its own store.
format: Optional response format (e.g. ``"json"``,
``"csv"``, ``"pdf"``). When provided, folded into
the OPA input as ``input.format`` so policies can
fork on it.
representation: Optional response representation name.
When provided, folded into the OPA input as
``input.representation``.
Returns:
The service's :class:`Decision`.
Raises:
OpaError: The service was unreachable, timed out, or
answered with a non-2xx status.
"""
document: dict[str, Any] = {
"subject": subject.as_input(),
"action": action,
"resource": resource.as_input(),
"roles": dict(roles),
"bindings": [binding.as_input() for binding in bindings],
}
if format is not None:
document["format"] = format
if representation is not None:
document["representation"] = representation
response = await self._post(self._decision_path, {"input": document})
return Decision.from_result(response.get("result", {}))
[docs]
async def check_many(
self,
*,
subject: Subject,
roles: Mapping[str, Any],
bindings: Sequence[RoleBinding],
items: Sequence[tuple[str, ResourceRef]],
format: str | None = None, # noqa: A002
representation: str | None = None,
) -> list[bool]:
"""Decide a batch of ``(action, resource)`` checks in one call.
Every item shares one *subject*, *roles*, and *bindings* --
the actions-envelope case: one user, many ``row x action``
checks for a whole list page. The service evaluates the
same per-one policy for each item, so a page costs a
single OPA round-trip rather than one call per item.
Args:
subject: The principal making the request.
roles: The role catalogue (see :meth:`check`).
bindings: The subject's role bindings.
items: The ``(action, resource)`` pairs to decide.
format: Optional response format (e.g. ``"json"``,
``"csv"``, ``"pdf"``). When provided, folded into
each query's input as ``input.format``.
representation: Optional response representation name.
When provided, folded into each query's input as
``input.representation``.
Returns:
One ``permit`` verdict per item, in the same order as
*items*. The bulk entrypoint returns the bottom-line
``permit`` only -- use :meth:`check` for the full
allow / deny breakdown.
Raises:
OpaError: The service was unreachable, timed out, or
answered with a non-2xx status.
"""
subject_input = subject.as_input()
roles_input = dict(roles)
bindings_input = [binding.as_input() for binding in bindings]
queries: list[dict[str, Any]] = []
for action, resource in items:
query: dict[str, Any] = {
"subject": subject_input,
"action": action,
"resource": resource.as_input(),
"roles": roles_input,
"bindings": bindings_input,
}
if format is not None:
query["format"] = format
if representation is not None:
query["representation"] = representation
queries.append(query)
response = await self._post(
self._bulk_path,
{"input": {"queries": queries}},
)
return _parse_bulk_result(response.get("result", []), len(items))
[docs]
async def compile_filter(
self,
*,
subject: Subject,
action: str,
resource_type: str,
roles: Mapping[str, Any],
bindings: Sequence[RoleBinding],
unknowns: Sequence[str] = ("input.resource.id",),
format: str | None = None, # noqa: A002
representation: str | None = None,
) -> FilterResult:
"""Compile the list policy into a row filter for *resource_type*.
Asks OPA to partially evaluate ``data.<opa_package>.allow``
with the resource left unknown, then translates the
residual into a :class:`FilterResult`. The resource
*type* is supplied as a known value so type-scoped and
global bindings resolve fully and only instance-level
constraints survive into the residual.
Args:
subject: The principal making the request.
action: The collection action, e.g. ``"task:list"``.
resource_type: The resource type being listed, e.g.
``"Task"``. Passed as a known value so the
residual is purely about instance fields.
roles: The role catalogue (see :meth:`check`).
bindings: The subject's role bindings.
unknowns: The ``input`` paths OPA treats as symbolic.
Defaults to ``("input.resource.id",)`` -- enough
for generic RBAC. Extend it (e.g. with
``"input.resource.created_by"``) when a custom
rule constrains other resource fields.
format: Optional response format (e.g. ``"json"``,
``"csv"``, ``"pdf"``). When provided, folded into
the OPA input as ``input.format`` so policies can
fork on it.
representation: Optional response representation name.
When provided, folded into the OPA input as
``input.representation``.
Returns:
The compiled :class:`FilterResult`.
Raises:
OpaError: The service failed, or returned a residual
the translator cannot represent as a SQL filter.
"""
input_doc: dict[str, Any] = {
"subject": subject.as_input(),
"action": action,
"resource": {"type": resource_type},
"roles": dict(roles),
"bindings": [binding.as_input() for binding in bindings],
}
if format is not None:
input_doc["format"] = format
if representation is not None:
input_doc["representation"] = representation
body = {
"query": f"data.{self._opa_package}.allow == true",
"input": input_doc,
"unknowns": list(unknowns),
}
response = await self._post("/v1/compile", body)
return _parse_compile_result(
response.get("result", {}),
opa_package=self._opa_package,
)
async def _post(
self,
path: str,
body: Mapping[str, Any],
) -> dict[str, Any]:
"""POST *body* as JSON to *path*, returning the parsed response.
Raises:
OpaError: The service was unreachable, timed out, or
answered with a non-2xx status.
"""
try:
response = await self._client.post(path, json=dict(body))
response.raise_for_status()
except httpx.HTTPError as exc:
msg = f"permissions service call failed: {exc}"
raise OpaError(msg) from exc
return response.json()
def _parse_bulk_result(result: Any, count: int) -> list[bool]:
"""Order a bulk ``decisions`` result back into a per-item list.
The bulk entrypoint returns a *set* of ``{index, permit}``
objects, so the response array has no guaranteed order;
``index`` ties each verdict to its query. An item with no
matching verdict defaults to ``False`` -- a partial answer is
never read as a silent allow.
"""
by_index: dict[int, bool] = {}
if isinstance(result, list):
for entry in result:
index = entry.get("index")
if isinstance(index, int):
by_index[index] = bool(entry.get("permit", False))
return [by_index.get(position, False) for position in range(count)]
# -- Compile-result translation -------------------------------------
#
# OPA's /v1/compile returns the residual as a disjunction of
# conjunctions of expressions (an AST). The helpers below walk
# that AST into a :class:`FilterResult`. Partial evaluation may
# factor ``default``-bearing rules into ``support`` -- a graph of
# generated rules referencing each other -- which the walker
# expands back into a flat OR-of-ANDs. The leaf expressions a
# resource-scoped RBAC residual produces are supported -- a
# comparison between an ``input.resource.<field>`` reference and a
# literal; anything else raises :class:`OpaError` rather than
# silently dropping a constraint.
#: OPA builtin name -> :class:`Condition` op. ``flip`` is the op
#: to use when the literal is on the left and the field reference
#: on the right (comparisons are not symmetric).
_OPERATORS: dict[str, tuple[str, str]] = {
"equal": ("eq", "eq"),
"eq": ("eq", "eq"),
"neq": ("ne", "ne"),
"lt": ("lt", "gt"),
"lte": ("le", "ge"),
"gt": ("gt", "lt"),
"gte": ("ge", "le"),
"internal.member_2": ("in", "in"),
}
#: Prefix every residual reference this translator understands
#: starts with. Generic RBAC list filtering leaves the resource
#: unknown, so every surviving constraint is on a resource field.
_RESOURCE_PREFIX = ("input", "resource")
#: A FilterResult that passes every row.
_ALWAYS_ALLOW = FilterResult(
always_allow=True,
always_deny=False,
conjunctions=(),
)
#: A FilterResult that rejects every row.
_ALWAYS_DENY = FilterResult(
always_allow=False,
always_deny=True,
conjunctions=(),
)
def _parse_compile_result(
result: Mapping[str, Any],
*,
opa_package: str | None = None,
) -> FilterResult:
"""Translate an OPA ``/v1/compile`` ``result`` into a FilterResult.
Handles the shapes OPA's partial evaluation produces:
* the flat form -- ``queries`` is an OR of conjunctions of
comparison expressions;
* the support form -- expressions reference generated support
rules (OPA factors every ``default``-bearing rule out into
``support``). A routing entrypoint yields a *graph*: support
rules referencing other support rules, conjoined with
existence checks on the consumer's rule packages. The graph
is expanded back into a flat OR-of-ANDs.
Args:
result: The ``result`` object of a ``/v1/compile`` response.
opa_package: The package the compiled rule lives in. When
given, a residual existence check on one of its rule
packages (``data.<opa_package>.rule...``) is accepted
as true -- see :meth:`_SupportWalker._ref_dnf`.
Raises:
OpaError: A residual expression the translator cannot
represent as a SQL filter.
"""
queries = result.get("queries")
# No `queries` key at all -- the query is unsatisfiable, so no
# row can ever pass.
if queries is None:
return _ALWAYS_DENY
walker = _SupportWalker(result.get("support") or [], opa_package)
disjuncts: list[tuple[Condition, ...]] = []
for query in queries:
disjuncts.extend(walker.query_dnf(query))
return _from_dnf(disjuncts)
def _from_dnf(disjuncts: list[tuple[Condition, ...]]) -> FilterResult:
"""Fold an OR-of-ANDs into a :class:`FilterResult`.
An empty conjunction is an unconditionally-true branch: if any
OR branch is free, the whole filter passes every row. No
branches at all means no row can ever pass.
"""
if any(not conjunction for conjunction in disjuncts):
return _ALWAYS_ALLOW
if not disjuncts:
return _ALWAYS_DENY
# The same conjunction can arrive through several support
# paths; keep the first of each. Deduped by equality, not
# hash -- an ``in`` condition holds an unhashable list.
unique: list[tuple[Condition, ...]] = []
for conjunction in disjuncts:
if conjunction not in unique:
unique.append(conjunction)
return FilterResult(
always_allow=False,
always_deny=False,
conjunctions=tuple(unique),
)
class _SupportWalker:
"""Expands a compile result's queries + support graph into DNF.
The DNF representation is a list of conjunctions (OR-of-ANDs):
an empty list is unsatisfiable, and an empty conjunction inside
the list is an unconditionally-true branch.
"""
def __init__(
self,
support: list[Mapping[str, Any]],
opa_package: str | None,
) -> None:
"""Index the support rules by their full dotted path."""
self._rules: dict[tuple[str, ...], list[Mapping[str, Any]]] = {}
for entry in support:
package_path = _support_package_path(entry)
for rule in entry.get("rules", []):
name = _support_rule_name(rule)
if name is None:
continue
key = (*package_path, name)
self._rules.setdefault(key, []).append(rule)
self._rule_package_prefix = (
("data", opa_package, "rule") if opa_package is not None else None
)
self._expanding: set[tuple[str, ...]] = set()
def query_dnf(
self,
exprs: list[Mapping[str, Any]],
) -> list[tuple[Condition, ...]]:
"""AND the expressions of one query body, returning DNF."""
acc: list[tuple[Condition, ...]] = [()]
for expr in exprs:
expr_disjuncts = self._expr_dnf(expr)
acc = [(*left, *right) for left in acc for right in expr_disjuncts]
return acc
def _expr_dnf(
self,
expr: Mapping[str, Any],
) -> list[tuple[Condition, ...]]:
"""Translate one expression into DNF."""
literal = _boolean_literal(expr)
if literal is not None:
if expr.get("negated"):
literal = not literal
return [()] if literal else []
terms = expr.get("terms")
# A single term is a bare reference (a support rule or an
# existence check), not a comparison call.
if isinstance(terms, dict):
return self._ref_dnf(expr, terms)
presence = self._presence_dnf(expr, terms)
if presence is not None:
return presence
return [(_parse_expr(expr),)]
def _presence_dnf(
self,
expr: Mapping[str, Any],
terms: Any,
) -> list[tuple[Condition, ...]] | None:
"""Resolve a rule-presence check, or ``None`` when not one.
The routing entrypoint's per-rule fall-through gates read
``data.<opa_package>.rule[...][...].allow != null`` -- true
exactly when the consumer file *defines* the rule. Partial
evaluation leaves the check in the residual only when the
rule is defined (with its value unresolved because it reads
unknowns); were it undefined, the ground-undefined reference
would have pruned the whole branch. OPA factors such a rule
into ``support``, so the reference resolves to an indexed
support rule here, and a rule declared with a ``default`` is
never undefined -- and never null -- at runtime. The check
is therefore constantly true (negated: dead).
"""
if not isinstance(terms, list) or len(terms) != _BINARY_TERMS:
return None
operator_term, left, right = terms
if (
operator_term.get("type") != "ref"
or _operator_name(operator_term) != "neq"
):
return None
if left.get("type") == "null":
ref_term = right
elif right.get("type") == "null":
ref_term = left
else:
return None
path = _ref_path(ref_term)
if path is None:
return None
rules = self._rules.get(tuple(path))
if rules is None:
return None
if not any(rule.get("default") for rule in rules):
# Without a default the rule is undefined at runtime
# whenever no clause fires, so whether it ``!= null``
# depends on the unknowns -- refuse rather than guess.
msg = (
f"presence check on support rule {'.'.join(path)!r} "
f"without a default clause"
)
raise OpaError(msg)
return [] if expr.get("negated") else [()]
def _ref_dnf(
self,
expr: Mapping[str, Any],
term: Mapping[str, Any],
) -> list[tuple[Condition, ...]]:
"""Translate a bare reference expression into DNF."""
path = _ref_path(term)
negated = bool(expr.get("negated", False))
if path is None:
msg = (
f"cannot translate residual expression {expr!r}: "
f"unsupported bare reference"
)
raise OpaError(msg)
key = tuple(path)
if key in self._rules:
if negated:
# ``not <support rule>`` resolves only when the rule
# expands to a constant -- the shape the routing
# entrypoint's presence-check gates produce (OPA
# factors ``not x != null`` into a ``__not...__``
# support rule). A genuinely conditional rule would
# need De Morgan over its clauses -- refuse rather
# than mistranslate.
expansion = self._rule_dnf(key)
if not expansion:
return [()]
if any(not clause for clause in expansion):
return []
msg = (
f"unsupported negated support-rule reference "
f"{'.'.join(path)!r}"
)
raise OpaError(msg)
return self._rule_dnf(key)
# An existence check on a consumer rule package
# (``data.<opa_package>.rule.<resource>...``): the routing
# entrypoint's most-specific-wins dispatch tests the rule
# package's document. Partial evaluation leaves the test
# in the residual -- positive or negated -- only when the
# package is loaded (with its rules unresolved because they
# read unknowns); were it absent, the reference would be
# ground-undefined and resolve away. Every rule file
# declares ``default allow``/``default deny`` so the loaded
# document is never empty: a positive check always holds at
# runtime and a negated one never does (a dead branch).
prefix = self._rule_package_prefix
if prefix is not None and key[: len(prefix)] == prefix:
return [] if negated else [()]
msg = f"unsupported residual reference {'.'.join(path)!r}"
raise OpaError(msg)
def _rule_dnf(self, key: tuple[str, ...]) -> list[tuple[Condition, ...]]:
"""Expand one support rule: the OR of its clause bodies."""
if key in self._expanding:
msg = f"circular support-rule reference at {'.'.join(key)!r}"
raise OpaError(msg)
self._expanding.add(key)
try:
disjuncts: list[tuple[Condition, ...]] = []
for rule in self._rules[key]:
if rule.get("default"):
# ``default x := false`` contributes nothing; a
# true default makes the rule unconditional.
head_value = rule.get("head", {}).get("value", {})
if head_value.get("value"):
return [()]
continue
disjuncts.extend(self.query_dnf(rule.get("body", [])))
return disjuncts
finally:
self._expanding.discard(key)
def _boolean_literal(expr: Mapping[str, Any]) -> bool | None:
"""Return the bool an expr is, when it is a bare boolean term.
OPA renders an always-true / always-false body expression as a
single boolean ``terms`` object (not a comparison call).
"""
terms = expr.get("terms")
if isinstance(terms, dict) and terms.get("type") == "boolean":
return bool(terms.get("value"))
return None
def _support_package_path(entry: Mapping[str, Any]) -> tuple[str, ...]:
"""Return the dotted path of a support package as a tuple."""
parts = entry.get("package", {}).get("path", [])
return tuple(part.get("value") for part in parts)
def _support_rule_name(rule: Mapping[str, Any]) -> str | None:
"""Return the name of a support rule, or ``None``."""
return rule.get("head", {}).get("name")
def _ref_path(term: Mapping[str, Any]) -> list[str] | None:
"""Return the dotted path of a ref term, or ``None``."""
if term.get("type") != "ref":
return None
return [part.get("value") for part in term.get("value", [])]
def _parse_expr(expr: Mapping[str, Any]) -> Condition:
"""Translate one residual expression into a :class:`Condition`.
Raises:
OpaError: The expression is not a binary comparison
between an ``input.resource`` field and a literal.
"""
terms = expr.get("terms")
negated = bool(expr.get("negated", False))
if not isinstance(terms, list) or len(terms) != _BINARY_TERMS:
msg = (
f"cannot translate residual expression {expr!r}: only "
f"binary comparisons of a resource field against a "
f"literal are supported"
)
raise OpaError(msg)
operator_term, left, right = terms
builtin = _operator_name(operator_term)
ops = _OPERATORS.get(builtin)
if ops is None:
msg = f"unsupported residual operator {builtin!r}"
raise OpaError(msg)
field, value, flipped = _split_operands(left, right)
if flipped and builtin == "internal.member_2":
msg = "unsupported residual: `in` with the field on the right"
raise OpaError(msg)
op = ops[1] if flipped else ops[0]
return Condition(field=field, op=op, value=value, negated=negated)
#: A binary-comparison expression has exactly three terms: the
#: operator reference and its two operands.
_BINARY_TERMS = 3
def _operator_name(term: Mapping[str, Any]) -> str:
"""Return the builtin name of an operator term (e.g. ``"equal"``).
Raises:
OpaError: The term is not an operator reference.
"""
if term.get("type") != "ref":
msg = f"expected an operator reference, got {term!r}"
raise OpaError(msg)
return ".".join(part["value"] for part in term["value"])
def _split_operands(
left: Mapping[str, Any],
right: Mapping[str, Any],
) -> tuple[str, Any, bool]:
"""Split a comparison's operands into ``(field, literal, flipped)``.
Exactly one operand must be an ``input.resource.<field>``
reference; the other must be a literal. ``flipped`` is
``True`` when the reference was the right operand.
Raises:
OpaError: Neither or both operands are resource
references, or the literal is not a scalar / array.
"""
left_field = _resource_field(left)
right_field = _resource_field(right)
if left_field is not None and right_field is None:
return left_field, _literal(right), False
if right_field is not None and left_field is None:
return right_field, _literal(left), True
msg = (
"cannot translate residual: a comparison must be between "
"exactly one input.resource field and one literal"
)
raise OpaError(msg)
def _resource_field(term: Mapping[str, Any]) -> str | None:
"""Return the field path of an ``input.resource`` ref, else ``None``.
``input.resource.id`` yields ``"id"``; a nested
``input.resource.owner.id`` yields ``"owner.id"``. A
reference to anything other than an ``input.resource`` field
yields ``None`` (it is treated as a literal-bearing operand by
the caller, which then rejects it).
"""
if term.get("type") != "ref":
return None
parts = term["value"]
head = tuple(part.get("value") for part in parts[: len(_RESOURCE_PREFIX)])
if head != _RESOURCE_PREFIX or len(parts) <= len(_RESOURCE_PREFIX):
return None
return ".".join(part["value"] for part in parts[len(_RESOURCE_PREFIX) :])
def _literal(term: Mapping[str, Any]) -> Any:
"""Return the Python value of a scalar / array / set term.
Raises:
OpaError: The term is not a literal the translator can
represent (e.g. another reference, or a call).
"""
kind = term.get("type")
if kind in {"string", "number", "boolean", "null"}:
return term.get("value")
if kind in {"array", "set"}:
return [_literal(element) for element in term["value"]]
msg = f"cannot translate residual operand of type {kind!r}"
raise OpaError(msg)