Policy¶

Policy answers one question, before anything external happens:

May this actor perform this proposed operation on these resources?

It is trusted configuration. The application writes it; the agent proposes work against it and can neither choose it, weaken it, nor name itself as someone else.

policy = gantry.Policy(
    name="data-agents",
    rules=[
        gantry.allow.query(sources=["analytics.*"]),
        gantry.allow.materialize(sources=["raw.*"], destinations=["agent_scratch.*"]),
        gantry.deny.materialize(destinations=["prod.*"]),
    ],
)

db = gantry.sql.connect("postgres", url=DATABASE_URL, policy=policy)

with gantry.actor.context(actor=gantry.actor.actor("agent", "etl-agent"), environment="prod"):
    run = await db.query(schemas=["analytics"])(sql)

run.admission.policy  # "data-agents"
run.admission.policy_hash  # "sha256:…"
run.admission.matched_rules  # ("allow-query-0",)

Policy decides before the engine is asked

A refused proposal is never submitted. The run exists — it is created before admission — and carries the decision, with no execution attached to it.

Defining a policy¶

gantry.Policy `dataclass` ¶

A named set of rules, and the hash of exactly those rules.

The version is derived from the rules rather than declared beside them, so a run that records sha256:… records the configuration that actually decided it. Two policies that differ anywhere hash differently; the same rules written in the same order hash the same in any process.

Source code in gantry/policy/model.py

@dataclass(frozen=True, slots=True)
class Policy:
    """A named set of rules, and the hash of exactly those rules.

    The version is derived from the rules rather than declared beside them, so
    a run that records `sha256:…` records the configuration that actually
    decided it. Two policies that differ anywhere hash differently; the same
    rules written in the same order hash the same in any process.
    """

    name: str
    rules: tuple[PolicyRule, ...] = ()

    def __init__(self, name: str, rules: Sequence[PolicyRule] = ()) -> None:
        if not isinstance(name, str) or not name.strip():
            raise PolicyConfigurationError("policy name must be a non-empty string")
        if isinstance(rules, PolicyRule):
            raise PolicyConfigurationError("rules must be a list of rules, not one rule")
        materialized = tuple(rules)
        for rule in materialized:
            if not isinstance(rule, PolicyRule):
                raise PolicyConfigurationError(f"rules must be PolicyRule objects, got {rule!r}")
        object.__setattr__(self, "name", name.strip())
        object.__setattr__(self, "rules", tuple(_named(materialized)))

    @property
    def version(self) -> str:
        """`sha256:…` over the canonical form of the name and rules."""
        canonical = json.dumps(
            {"name": self.name, "rules": [rule.as_dict() for rule in self.rules]},
            sort_keys=True,
            separators=(",", ":"),
        )
        return f"sha256:{sha256(canonical.encode()).hexdigest()}"

    def as_dict(self) -> dict[str, object]:
        return {
            "name": self.name,
            "version": self.version,
            "rules": [rule.as_dict() for rule in self.rules],
        }

version `property` ¶

version: str

sha256:… over the canonical form of the name and rules.

gantry.PolicyRule `dataclass` ¶

One rule. Every dimension left as None is unconstrained, except one.

destinations is the exception, and the asymmetry is deliberate: an allow rule that does not name destinations authorizes no writes at all. Reading the wrong table is a leak; writing the wrong table destroys something, so write authority is never granted by omission. deny rules keep the plain reading — a dimension left out simply does not narrow the rule.

constraints are ceilings the request must already be under. A rule cannot raise a limit the operation configured, so trusted constraints only ever compose toward less authority.

require_confirmation does not change the answer — the rule still allows — it asks the host to tell the user before the work happens. Authority and "should someone be told" are separate questions, and collapsing them would turn every sensitive operation into a refusal.

Source code in gantry/policy/model.py

@dataclass(frozen=True, slots=True)
class PolicyRule:
    """One rule. Every dimension left as `None` is unconstrained, except one.

    `destinations` is the exception, and the asymmetry is deliberate: an allow
    rule that does not name destinations authorizes no writes at all. Reading
    the wrong table is a leak; writing the wrong table destroys something, so
    write authority is never granted by omission. `deny` rules keep the plain
    reading — a dimension left out simply does not narrow the rule.

    `constraints` are ceilings the request must already be under. A rule cannot
    raise a limit the operation configured, so trusted constraints only ever
    compose toward less authority.

    `require_confirmation` does not change the answer — the rule still allows —
    it asks the host to tell the user before the work happens. Authority and
    "should someone be told" are separate questions, and collapsing them would
    turn every sensitive operation into a refusal.
    """

    effect: Effect
    name: str | None = None
    require_confirmation: bool = False
    confirmation_code: ConfirmationReasonCode | str | None = None
    confirmation_message: str | None = None
    actors: tuple[str, ...] | None = None
    operations: tuple[OperationKind, ...] | None = None
    engines: tuple[str, ...] | None = None
    sources: tuple[str, ...] | None = None
    destinations: tuple[str, ...] | None = None
    environments: tuple[str, ...] | None = None
    constraints: Mapping[str, float] = field(default_factory=dict)

    def __post_init__(self) -> None:
        object.__setattr__(self, "effect", _effect(self.effect))
        if self.name is not None and not self.name.strip():
            raise PolicyConfigurationError("rule name must not be empty when given")
        object.__setattr__(self, "actors", _names(self.actors, "actors"))
        object.__setattr__(self, "engines", _names(self.engines, "engines"))
        object.__setattr__(self, "environments", _names(self.environments, "environments"))
        object.__setattr__(self, "sources", _patterns(self.sources, "sources"))
        object.__setattr__(self, "destinations", _patterns(self.destinations, "destinations"))
        object.__setattr__(self, "operations", _operations(self.operations))
        object.__setattr__(self, "constraints", _constraints(self.constraints))
        self._check_confirmation()

    def _check_confirmation(self) -> None:
        """A confirmation prompt nobody will ever see is a mistake, not a preference."""
        if self.confirmation_code is not None:
            try:
                object.__setattr__(
                    self, "confirmation_code", ConfirmationReasonCode(self.confirmation_code)
                )
            except ValueError as error:
                known = ", ".join(code.value for code in ConfirmationReasonCode)
                raise PolicyConfigurationError(
                    f"unknown confirmation code: {self.confirmation_code!r}; v0 has {known}"
                ) from error
        if self.confirmation_message is not None and not self.confirmation_message.strip():
            raise PolicyConfigurationError("confirmation message must not be empty when given")
        if self.require_confirmation and self.effect is Effect.DENY:
            raise PolicyConfigurationError(
                "a deny rule cannot require confirmation: nothing it matches will run"
            )
        if not self.require_confirmation and (
            self.confirmation_code is not None or self.confirmation_message is not None
        ):
            raise PolicyConfigurationError(
                "confirmation_code and confirmation_message need require_confirmation=True, "
                "or nothing will ever show them"
            )

    def as_dict(self) -> dict[str, object]:
        payload: dict[str, object] = {"effect": self.effect.value}
        if self.name is not None:
            payload["name"] = self.name
        if self.require_confirmation:
            payload["require_confirmation"] = True
        if self.confirmation_code is not None:
            payload["confirmation_code"] = str(self.confirmation_code)
        if self.confirmation_message is not None:
            payload["confirmation_message"] = self.confirmation_message
        for key in ("actors", "engines", "sources", "destinations", "environments"):
            value = getattr(self, key)
            if value is not None:
                payload[key] = list(value)
        if self.operations is not None:
            payload["operations"] = [operation.value for operation in self.operations]
        if self.constraints:
            payload["constraints"] = dict(sorted(self.constraints.items()))
        return payload

Allow rules¶

Rules that grant authority. Omitting sources allows any source to be read; omitting destinations grants no write authority at all, so the writing operations require them.

gantry.policy.allow.query ¶

query(
    *,
    name: str | None = None,
    require_confirmation: bool = False,
    confirmation_code: ConfirmationReasonCode
    | str
    | None = None,
    confirmation_message: str | None = None,
    actors: Sequence[str] | None = None,
    engines: Sequence[str] | None = None,
    sources: Sequence[str] | None = None,
    destinations: Sequence[str] | None = None,
    environments: Sequence[str] | None = None,
    constraints: Mapping[str, float] | None = None,
) -> PolicyRule

Allow reading. Omitting sources allows any source to be read.

destinations is for the one case where a query writes — a pipeline ending in $out, an INSERT submitted through a query operation. Omitting it grants no write authority, so a read policy stays a read policy.

Source code in gantry/policy/allow.py

def query(
    *,
    name: str | None = None,
    require_confirmation: bool = False,
    confirmation_code: ConfirmationReasonCode | str | None = None,
    confirmation_message: str | None = None,
    actors: Sequence[str] | None = None,
    engines: Sequence[str] | None = None,
    sources: Sequence[str] | None = None,
    destinations: Sequence[str] | None = None,
    environments: Sequence[str] | None = None,
    constraints: Mapping[str, float] | None = None,
) -> PolicyRule:
    """Allow reading. Omitting `sources` allows any source to be read.

    `destinations` is for the one case where a query writes — a pipeline ending
    in `$out`, an `INSERT` submitted through a query operation. Omitting it
    grants no write authority, so a read policy stays a read policy.
    """
    return _allow(
        OperationKind.QUERY,
        name=name,
        require_confirmation=require_confirmation,
        confirmation_code=confirmation_code,
        confirmation_message=confirmation_message,
        actors=actors,
        engines=engines,
        sources=sources,
        destinations=destinations,
        environments=environments,
        constraints=constraints,
    )

gantry.policy.allow.materialize ¶

materialize(
    *,
    destinations: Sequence[str],
    name: str | None = None,
    require_confirmation: bool = False,
    confirmation_code: ConfirmationReasonCode
    | str
    | None = None,
    confirmation_message: str | None = None,
    actors: Sequence[str] | None = None,
    engines: Sequence[str] | None = None,
    sources: Sequence[str] | None = None,
    environments: Sequence[str] | None = None,
    constraints: Mapping[str, float] | None = None,
) -> PolicyRule

Allow a materialization that lands in destinations.

Source code in gantry/policy/allow.py

def materialize(
    *,
    destinations: Sequence[str],
    name: str | None = None,
    require_confirmation: bool = False,
    confirmation_code: ConfirmationReasonCode | str | None = None,
    confirmation_message: str | None = None,
    actors: Sequence[str] | None = None,
    engines: Sequence[str] | None = None,
    sources: Sequence[str] | None = None,
    environments: Sequence[str] | None = None,
    constraints: Mapping[str, float] | None = None,
) -> PolicyRule:
    """Allow a materialization that lands in `destinations`."""
    return _allow(
        OperationKind.MATERIALIZE,
        name=name,
        require_confirmation=require_confirmation,
        confirmation_code=confirmation_code,
        confirmation_message=confirmation_message,
        actors=actors,
        engines=engines,
        sources=sources,
        destinations=destinations,
        environments=environments,
        constraints=constraints,
    )

gantry.policy.allow.batch ¶

batch(
    *,
    destinations: Sequence[str],
    name: str | None = None,
    require_confirmation: bool = False,
    confirmation_code: ConfirmationReasonCode
    | str
    | None = None,
    confirmation_message: str | None = None,
    actors: Sequence[str] | None = None,
    engines: Sequence[str] | None = None,
    sources: Sequence[str] | None = None,
    environments: Sequence[str] | None = None,
    constraints: Mapping[str, float] | None = None,
) -> PolicyRule

Allow a batch job that writes into destinations.

Source code in gantry/policy/allow.py

def batch(
    *,
    destinations: Sequence[str],
    name: str | None = None,
    require_confirmation: bool = False,
    confirmation_code: ConfirmationReasonCode | str | None = None,
    confirmation_message: str | None = None,
    actors: Sequence[str] | None = None,
    engines: Sequence[str] | None = None,
    sources: Sequence[str] | None = None,
    environments: Sequence[str] | None = None,
    constraints: Mapping[str, float] | None = None,
) -> PolicyRule:
    """Allow a batch job that writes into `destinations`."""
    return _allow(
        OperationKind.BATCH,
        name=name,
        require_confirmation=require_confirmation,
        confirmation_code=confirmation_code,
        confirmation_message=confirmation_message,
        actors=actors,
        engines=engines,
        sources=sources,
        destinations=destinations,
        environments=environments,
        constraints=constraints,
    )

gantry.policy.allow.stream ¶

stream(
    *,
    destinations: Sequence[str],
    name: str | None = None,
    require_confirmation: bool = False,
    confirmation_code: ConfirmationReasonCode
    | str
    | None = None,
    confirmation_message: str | None = None,
    actors: Sequence[str] | None = None,
    engines: Sequence[str] | None = None,
    sources: Sequence[str] | None = None,
    environments: Sequence[str] | None = None,
    constraints: Mapping[str, float] | None = None,
) -> PolicyRule

Allow submitting a streaming job that writes into destinations.

Source code in gantry/policy/allow.py

def stream(
    *,
    destinations: Sequence[str],
    name: str | None = None,
    require_confirmation: bool = False,
    confirmation_code: ConfirmationReasonCode | str | None = None,
    confirmation_message: str | None = None,
    actors: Sequence[str] | None = None,
    engines: Sequence[str] | None = None,
    sources: Sequence[str] | None = None,
    environments: Sequence[str] | None = None,
    constraints: Mapping[str, float] | None = None,
) -> PolicyRule:
    """Allow submitting a streaming job that writes into `destinations`."""
    return _allow(
        OperationKind.STREAM,
        name=name,
        require_confirmation=require_confirmation,
        confirmation_code=confirmation_code,
        confirmation_message=confirmation_message,
        actors=actors,
        engines=engines,
        sources=sources,
        destinations=destinations,
        environments=environments,
        constraints=constraints,
    )

Deny rules¶

A matching deny wins over every allow.