Write scope

write_scope

WriteAccessDecision dataclass

Verdict for one attempted filesystem mutation.

Source code in packages/axm/src/axm/tools/write_scope.py

@dataclass(frozen=True)
class WriteAccessDecision:
    """Verdict for one attempted filesystem mutation."""

    allowed: bool
    reason: str
    resolved_location: str | None = None
    consulted_prefixes: tuple[str, ...] = ()

WriteContract dataclass

Validated wire representation of a session filesystem write scope.

Source code in packages/axm/src/axm/tools/write_scope.py

@dataclass(frozen=True)
class WriteContract:
    """Validated wire representation of a session filesystem write scope."""

    execution_root: str
    allowed_prefixes: tuple[str, ...] = ()

    @classmethod
    def from_mapping(cls, raw: Mapping[str, object]) -> WriteContract:
        """Validate and normalize a transported contract mapping."""
        raw_root = raw.get("execution_root")
        if not isinstance(raw_root, str) or not raw_root.strip():
            raise ValueError("execution_root must be a non-empty string")
        root = os.path.realpath(raw_root.strip())
        raw_prefixes = raw.get("allowed_prefixes", ())
        if isinstance(raw_prefixes, str) or not isinstance(raw_prefixes, Sequence):
            raise ValueError("allowed_prefixes must be a sequence of strings")
        if any(not isinstance(prefix, str) for prefix in raw_prefixes):
            raise ValueError("allowed_prefixes must contain only strings")
        prefixes = tuple(
            dict.fromkeys(
                _absolute_location(root, prefix)
                for prefix in raw_prefixes
                if isinstance(prefix, str)
            )
        )
        return cls(execution_root=root, allowed_prefixes=prefixes)

    @classmethod
    def from_json(cls, raw: str) -> WriteContract:
        """Decode, validate and normalize a JSON transport payload."""
        decoded = json.loads(raw)
        if not isinstance(decoded, Mapping):
            raise ValueError("write contract must be a JSON object")
        return cls.from_mapping(decoded)

    def resolve(self, base: str, candidate: str) -> str:
        """Resolve a candidate location relative to an explicit base."""
        return _absolute_location(base, candidate)

    def permits(self, location: str) -> bool:
        """Return whether a resolved location is under an allowed prefix."""
        return any(
            location == prefix or location.startswith(prefix + os.sep)
            for prefix in self.allowed_prefixes
        )

    def contains(self, location: str) -> bool:
        """Return whether a resolved location is under the execution root."""
        root = self.execution_root
        return location == root or location.startswith(root + os.sep)

contains(location)

Return whether a resolved location is under the execution root.

Source code in packages/axm/src/axm/tools/write_scope.py

def contains(self, location: str) -> bool:
    """Return whether a resolved location is under the execution root."""
    root = self.execution_root
    return location == root or location.startswith(root + os.sep)

from_json(raw) classmethod

Decode, validate and normalize a JSON transport payload.

Source code in packages/axm/src/axm/tools/write_scope.py

@classmethod
def from_json(cls, raw: str) -> WriteContract:
    """Decode, validate and normalize a JSON transport payload."""
    decoded = json.loads(raw)
    if not isinstance(decoded, Mapping):
        raise ValueError("write contract must be a JSON object")
    return cls.from_mapping(decoded)

from_mapping(raw) classmethod

Validate and normalize a transported contract mapping.

Source code in packages/axm/src/axm/tools/write_scope.py

@classmethod
def from_mapping(cls, raw: Mapping[str, object]) -> WriteContract:
    """Validate and normalize a transported contract mapping."""
    raw_root = raw.get("execution_root")
    if not isinstance(raw_root, str) or not raw_root.strip():
        raise ValueError("execution_root must be a non-empty string")
    root = os.path.realpath(raw_root.strip())
    raw_prefixes = raw.get("allowed_prefixes", ())
    if isinstance(raw_prefixes, str) or not isinstance(raw_prefixes, Sequence):
        raise ValueError("allowed_prefixes must be a sequence of strings")
    if any(not isinstance(prefix, str) for prefix in raw_prefixes):
        raise ValueError("allowed_prefixes must contain only strings")
    prefixes = tuple(
        dict.fromkeys(
            _absolute_location(root, prefix)
            for prefix in raw_prefixes
            if isinstance(prefix, str)
        )
    )
    return cls(execution_root=root, allowed_prefixes=prefixes)

permits(location)

Return whether a resolved location is under an allowed prefix.

Source code in packages/axm/src/axm/tools/write_scope.py

def permits(self, location: str) -> bool:
    """Return whether a resolved location is under an allowed prefix."""
    return any(
        location == prefix or location.startswith(prefix + os.sep)
        for prefix in self.allowed_prefixes
    )

resolve(base, candidate)

Resolve a candidate location relative to an explicit base.

Source code in packages/axm/src/axm/tools/write_scope.py

def resolve(self, base: str, candidate: str) -> str:
    """Resolve a candidate location relative to an explicit base."""
    return _absolute_location(base, candidate)

decide_write_access(contract, tool_name, tool_input=None)

Decide whether an AXM call may write where its payload requests.

Source code in packages/axm/src/axm/tools/write_scope.py

def decide_write_access(
    contract: WriteContract | Mapping[str, object] | None,
    tool_name: str,
    tool_input: Mapping[str, object] | None = None,
) -> WriteAccessDecision:
    """Decide whether an AXM call may write where its payload requests."""
    if contract is None:
        return WriteAccessDecision(allowed=True, reason="no write contract is in force")
    resolved_contract = (
        contract
        if isinstance(contract, WriteContract)
        else WriteContract.from_mapping(contract)
    )
    payload: Mapping[str, object] = tool_input or {}
    canonical, payload, facade_error = _canonical_call(tool_name, payload)
    if facade_error is not None:
        return WriteAccessDecision(
            allowed=False,
            reason=facade_error,
            consulted_prefixes=resolved_contract.allowed_prefixes,
        )
    shape = _MUTATION_TOOLS.get(canonical)
    if shape is None:
        return _unclassified_tool_decision(canonical, payload, resolved_contract)
    return _decide_declared_mutation(
        canonical,
        payload,
        shape,
        resolved_contract,
    )

write_contract_from_env(env=None)

Load the optional process-scoped contract, failing closed if malformed.

Source code in packages/axm/src/axm/tools/write_scope.py

def write_contract_from_env(
    env: Mapping[str, str] | None = None,
) -> WriteContract | None:
    """Load the optional process-scoped contract, failing closed if malformed."""
    source = os.environ if env is None else env
    raw = source.get(WRITE_CONTRACT_ENV)
    if raw is None:
        return None
    try:
        return WriteContract.from_json(raw)
    except (TypeError, ValueError, json.JSONDecodeError) as exc:
        raise ValueError(f"invalid {WRITE_CONTRACT_ENV}: {exc}") from exc