orchestrata_ultimate.kernel.pam.enforcer

Reference › orchestrata_ultimate › kernel › pam › enforcer

The PAM++ enforcement engine, SPEC-001 story US-003 (REQ-001-US-003-011 to -020). PAMEnforcer answers whether a volk may act on a resource, by asking Prolog rules in the pam module. Its answers are the typed results from types, and the pam package re-exports it. The Prolog bridge is passed in as an IGovernancePrologBridge, and an optional graph audit sink records violations. It is the volk-level counterpart of the message-level PolicyEvaluator.

validate_authority(sender, action, target) asks pam_owns/2; if the sender does not own the target it checks for delegated authority, and otherwise it logs an authority_denied violation and returns a denied AuthorityResult naming the actual owner.
check_ownership(volk, component) asks pam_owns/2, then looks for a temporary ownership grant that has not expired, and returns an OwnershipResult that is valid or invalid.
enforce_boundary(from_volk, to_volk, resource) asks cross_boundary_allowed/3; when allowed it also checks boundary_exception/3, and when denied it logs a boundary_violation.
check_permission(volk, resource, operation) asks has_permission/3 and, if granted, grant_active/3 to tell a permanent grant from a temporary one with an expiry; a denial logs permission_denied.
delegate_ownership and grant_temporary_permission assert time-limited facts into Prolog, after checking the caller really owns the resource.

Every Prolog call runs in an executor, and a failed query is logged and treated as a denial. _log_violation keeps violations in memory, truncating the list to a retained count once it passes a maximum, logs a warning and writes to the graph sink if one exists; get_violations returns the list.

At a glance


Source	`src/orchestrata_ultimate/kernel/pam/enforcer.py`
Lines of code	592
Classes	1
Functions	0
Imports (internal)	`orchestrata_ultimate.core.base.traces`, `orchestrata_ultimate.core.enums`, `orchestrata_ultimate.core.protocols`, `orchestrata_ultimate.kernel.pam.types`
Imported by	`orchestrata_ultimate.kernel.pam`
Tests that import it	none

Class diagram

Dependency diagram

API

Classes

`PAMEnforcer`

Decorators: traces(spec='001', story='US-003', reqs=['REQ-001-US-003-011', 'REQ-001-US-003-012', 'REQ-001-US-003-013', 'REQ-001-US-… · Line 49

PAM++ Enforcement Engine.

Method	Signature	Summary
`__init__`	`__init__(prolog_bridge: IGovernancePrologBridge, graph_audit: Any \| None = None, max_violations_before_truncate: int = MAX_VIOLATIONS_BEFORE_TRUNCATE, violation_log_retain_count: int = VIOLATION_LOG_RETAIN_COUNT) -> None`	Initialize PAM enforcer.
`_check_direct_ownership`	`async _check_direct_ownership(loop: Any, sender: str, target: str) -> bool`	Query Prolog to check if sender directly owns target.
`_check_delegation`	`async _check_delegation(loop: Any, sender: str, target: str) -> dict \| None`	Query Prolog to check if sender has delegated authority over target.
`_lookup_actual_owner`	`async _lookup_actual_owner(loop: Any, target: str) -> str \| None`	Query Prolog to find the actual owner of a target resource.
`validate_authority`	`async validate_authority(sender: str, action: str, target: str) -> AuthorityResult`	Validate sender has authority to perform action (REQ-001-US-003-011).
`_check_temporary_ownership`	`async _check_temporary_ownership(loop: Any, volk: str, component: str, execution_time: float) -> OwnershipResult \| None`	Query Prolog for temporary/delegated ownership and return a result if found.
`_lookup_actual_component_owner`	`async _lookup_actual_component_owner(loop: Any, component: str) -> str \| None`	Query Prolog to find the actual owner of a component.
`check_ownership`	`async check_ownership(volk: str, component: str) -> OwnershipResult`	Check component ownership (REQ-001-US-003-012, REQ-001-US-003-013).
`enforce_boundary`	`async enforce_boundary(from_volk: str, to_volk: str, resource: str) -> BoundaryResult`	Enforce boundary crossing rules (REQ-001-US-003-014, REQ-001-US-003-015).
`check_permission`	`async check_permission(volk: str, resource: str, operation: str) -> PermissionResult`	Check permission for operation (REQ-001-US-003-016, REQ-001-US-003-017).
`delegate_ownership`	`async delegate_ownership(owner: str, delegate: str, resource: str, duration: int) -> bool`	Delegate ownership temporarily (REQ-001-US-003-018).
`grant_temporary_permission`	`async grant_temporary_permission(volk: str, permission: str, duration: int) -> bool`	Grant temporary permission (REQ-001-US-003-019).
`_log_violation`	`async _log_violation(violation_type: str, details: dict[str, Any]) -> None`	Log permission violation (REQ-001-US-003-020).
`get_violations`	`get_violations() -> list[dict[str, Any]]`	Get all logged violations.

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/kernel/pam/enforcer.py

# mypy: ignore-errors
"""PAM++ Enforcement Engine (PAMEnforcer).

Decomposed from kernel/pam_enforcer.py. Validates authority, ownership,
boundaries, and permissions using Prolog rules from pam_rules.pl.
"""

import asyncio
import logging
import time
from datetime import UTC, datetime, timedelta
from typing import Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.protocols import IGovernancePrologBridge
from orchestrata_ultimate.kernel.pam.types import (
    MAX_VIOLATIONS_BEFORE_TRUNCATE,
    VIOLATION_LOG_RETAIN_COUNT,
    AuthorityResult,
    AuthorityStatus,
    BoundaryResult,
    BoundaryStatus,
    OwnershipResult,
    OwnershipStatus,
    PermissionResult,
    PermissionStatus,
    _utc_now,
)

logger = logging.getLogger(__name__)


@traces(
    spec="001",
    story="US-003",
    reqs=[
        "REQ-001-US-003-011",
        "REQ-001-US-003-012",
        "REQ-001-US-003-013",
        "REQ-001-US-003-014",
        "REQ-001-US-003-015",
        "REQ-001-US-003-016",
        "REQ-001-US-003-017",
        "REQ-001-US-003-018",
        "REQ-001-US-003-019",
        "REQ-001-US-003-020",
    ],
)
class PAMEnforcer:
    """PAM++ Enforcement Engine.

    Validates authority, ownership, boundaries, and permissions
    using Prolog rules from pam_rules.pl module.
    """

    def __init__(
        self,
        prolog_bridge: IGovernancePrologBridge,
        graph_audit: Any | None = None,
        max_violations_before_truncate: int = MAX_VIOLATIONS_BEFORE_TRUNCATE,
        violation_log_retain_count: int = VIOLATION_LOG_RETAIN_COUNT,
    ) -> None:
        """Initialize PAM enforcer.

        Args:
            prolog_bridge: IGovernancePrologBridge for rule queries
            graph_audit: Graph audit trail (optional)
            max_violations_before_truncate: Max violations before truncation.
            violation_log_retain_count: Number of violations to retain after truncation.
        """
        self._prolog = prolog_bridge
        self._graph = graph_audit
        self._violations: list[dict[str, Any]] = []
        self._max_violations_before_truncate = max_violations_before_truncate
        self._violation_log_retain_count = violation_log_retain_count

    @traces(spec="001", story="US-003", reqs=["REQ-001-US-003-011"])
    async def _check_direct_ownership(
        self, loop: Any, sender: str, target: str
    ) -> bool:
        """Query Prolog to check if sender directly owns target.

        Args:
            loop: Running event loop for executor dispatch
            sender: Volk to check ownership for
            target: Target resource

        Returns:
            True if sender owns target, False otherwise.
        """
        query = f"pam:pam_owns('{sender}', '{target}')"
        try:
            return await loop.run_in_executor(None, self._prolog.query_bool, query)
        except Exception as e:
            logger.error(f"Prolog query failed for ownership check: {e}")
            return False

    @traces(spec="001", story="US-003", reqs=["REQ-001-US-003-018"])
    async def _check_delegation(
        self, loop: Any, sender: str, target: str
    ) -> dict | None:
        """Query Prolog to check if sender has delegated authority over target.

        Args:
            loop: Running event loop for executor dispatch
            sender: Volk to check delegation for
            target: Target resource

        Returns:
            Delegation dict with Owner key if active, None otherwise.
        """
        query = f"pam:delegation_active(Owner, '{sender}', '{target}')"
        try:
            return await loop.run_in_executor(None, self._prolog.query_one, query)
        except Exception as e:
            logger.error(f"Prolog query failed for delegation check: {e}")
            return None

    @traces(spec="001", story="US-003", reqs=["REQ-001-US-003-011"])
    async def _lookup_actual_owner(self, loop: Any, target: str) -> str | None:
        """Query Prolog to find the actual owner of a target resource.

        Args:
            loop: Running event loop for executor dispatch
            target: Target resource to look up

        Returns:
            Owner alias string, or None if not found.
        """
        owner_query = f"pam:pam_owns(Owner, '{target}')"
        try:
            owner_result = await loop.run_in_executor(
                None, self._prolog.query_one, owner_query
            )
            return owner_result.get("Owner") if owner_result else None
        except Exception as e:
            logger.error(f"Prolog query failed for owner lookup: {e}")
            return None

    @traces(spec="001", story="US-003", reqs=["REQ-001-US-003-011"])
    async def validate_authority(
        self, sender: str, action: str, target: str
    ) -> AuthorityResult:
        """Validate sender has authority to perform action (REQ-001-US-003-011).

        Args:
            sender: Volk performing action
            action: Action being performed
            target: Target of action

        Returns:
            AuthorityResult with validation status
        """
        start_time = time.time()
        loop = asyncio.get_running_loop()

        owns = await self._check_direct_ownership(loop, sender, target)
        execution_time = (time.time() - start_time) * 1000

        if owns:
            return AuthorityResult(
                status=AuthorityStatus.AUTHORIZED,
                sender=sender,
                action=action,
                target=target,
                reason=f"{sender} owns {target}",
                execution_time_ms=execution_time,
            )

        delegation = await self._check_delegation(loop, sender, target)
        if delegation:
            return AuthorityResult(
                status=AuthorityStatus.AUTHORIZED,
                sender=sender,
                action=action,
                target=target,
                reason=f"{sender} has delegated authority from {delegation.get('Owner')}",
                required_authority=delegation.get("Owner"),
                execution_time_ms=execution_time,
            )

        actual_owner = await self._lookup_actual_owner(loop, target)

        # Log violation (REQ-001-US-003-020)
        await self._log_violation(
            "authority_denied",
            {
                "sender": sender,
                "action": action,
                "target": target,
                "actual_owner": actual_owner,
            },
        )

        return AuthorityResult(
            status=AuthorityStatus.DENIED,
            sender=sender,
            action=action,
            target=target,
            reason=f"{sender} does not own {target}",
            required_authority=actual_owner,
            execution_time_ms=execution_time,
        )

    @traces(spec="001", story="US-003", reqs=["REQ-001-US-003-018"])
    async def _check_temporary_ownership(
        self, loop: Any, volk: str, component: str, execution_time: float
    ) -> OwnershipResult | None:
        """Query Prolog for temporary/delegated ownership and return a result if found.

        Args:
            loop: Running event loop for executor dispatch
            volk: Volk to check delegation for
            component: Component being accessed
            execution_time: Elapsed time in ms for the result object

        Returns:
            OwnershipResult with DELEGATED or EXPIRED status if delegation exists,
            None if no delegation found.
        """
        delegation_query = f"pam:temporary_ownership('{volk}', '{component}', ExpiresAt, OriginalOwner)"
        try:
            delegation = await loop.run_in_executor(
                None, self._prolog.query_one, delegation_query
            )
        except Exception as e:
            logger.error(f"Prolog query failed for delegation check: {e}")
            delegation = None

        if not delegation:
            return None

        expires_at = datetime.fromtimestamp(delegation.get("ExpiresAt", 0), tz=UTC)
        if expires_at > _utc_now():
            return OwnershipResult(
                status=OwnershipStatus.DELEGATED,
                volk=volk,
                component=component,
                reason=f"{volk} has delegated ownership until {expires_at}",
                delegated_from=delegation.get("OriginalOwner"),
                expires_at=expires_at,
                execution_time_ms=execution_time,
            )
        return OwnershipResult(
            status=OwnershipStatus.EXPIRED,
            volk=volk,
            component=component,
            reason=f"Delegation expired at {expires_at}",
            delegated_from=delegation.get("OriginalOwner"),
            expires_at=expires_at,
            execution_time_ms=execution_time,
        )

    @traces(spec="001", story="US-003", reqs=["REQ-001-US-003-012"])
    async def _lookup_actual_component_owner(
        self, loop: Any, component: str
    ) -> str | None:
        """Query Prolog to find the actual owner of a component.

        Args:
            loop: Running event loop for executor dispatch
            component: Component to look up

        Returns:
            Actual owner alias string, or None if not found.
        """
        owner_query = f"pam:pam_owns(ActualOwner, '{component}')"
        try:
            owner_result = await loop.run_in_executor(
                None, self._prolog.query_one, owner_query
            )
            return owner_result.get("ActualOwner") if owner_result else None
        except Exception as e:
            logger.error(f"Prolog query failed for owner lookup: {e}")
            return None

    @traces(spec="001", story="US-003", reqs=["REQ-001-US-003-012"])
    async def check_ownership(self, volk: str, component: str) -> OwnershipResult:
        """Check component ownership (REQ-001-US-003-012, REQ-001-US-003-013).

        Args:
            volk: Volk to check
            component: Component being accessed

        Returns:
            OwnershipResult with ownership status
        """
        start_time = time.time()
        loop = asyncio.get_running_loop()

        # Check direct ownership (async)
        query = f"pam:pam_owns('{volk}', '{component}')"
        try:
            owns = await loop.run_in_executor(None, self._prolog.query_bool, query)
        except Exception as e:
            logger.error(f"Prolog query failed for ownership check: {e}")
            owns = False

        execution_time = (time.time() - start_time) * 1000

        if owns:
            return OwnershipResult(
                status=OwnershipStatus.VALID,
                volk=volk,
                component=component,
                reason=f"{volk} owns {component}",
                execution_time_ms=execution_time,
            )

        delegation_result = await self._check_temporary_ownership(
            loop, volk, component, execution_time
        )
        if delegation_result is not None:
            return delegation_result

        actual_owner = await self._lookup_actual_component_owner(loop, component)

        return OwnershipResult(
            status=OwnershipStatus.INVALID,
            volk=volk,
            component=component,
            reason=f"{volk} does not own {component}",
            actual_owner=actual_owner,
            execution_time_ms=execution_time,
        )

    @traces(
        spec="001",
        story="US-003",
        reqs=[
            "REQ-001-US-003-013",
            "REQ-001-US-003-014",
            "REQ-001-US-003-015",
        ],
    )
    async def enforce_boundary(
        self, from_volk: str, to_volk: str, resource: str
    ) -> BoundaryResult:
        """Enforce boundary crossing rules (REQ-001-US-003-014, REQ-001-US-003-015).

        Args:
            from_volk: Source volk
            to_volk: Target volk
            resource: Resource being accessed

        Returns:
            BoundaryResult with enforcement decision
        """
        start_time = time.time()
        loop = asyncio.get_running_loop()

        # Check if cross-boundary allowed (async)
        query = f"pam:cross_boundary_allowed('{from_volk}', '{to_volk}', '{resource}')"
        try:
            allowed = await loop.run_in_executor(None, self._prolog.query_bool, query)
        except Exception as e:
            logger.error(f"Prolog query failed for boundary check: {e}")
            allowed = False

        execution_time = (time.time() - start_time) * 1000

        if allowed:
            # Check if exception granted (async)
            exception_query = (
                f"pam:boundary_exception('{from_volk}', '{to_volk}', '{resource}')"
            )
            try:
                has_exception = await loop.run_in_executor(
                    None, self._prolog.query_bool, exception_query
                )
            except Exception as e:
                logger.error(f"Prolog query failed for exception check: {e}")
                has_exception = False

            return BoundaryResult(
                status=BoundaryStatus.ALLOWED,
                from_volk=from_volk,
                to_volk=to_volk,
                resource=resource,
                reason="Cross-boundary access allowed",
                exception_granted=has_exception,
                execution_time_ms=execution_time,
            )

        # Log violation (REQ-001-US-003-020)
        await self._log_violation(
            "boundary_violation",
            {"from_volk": from_volk, "to_volk": to_volk, "resource": resource},
        )

        return BoundaryResult(
            status=BoundaryStatus.DENIED,
            from_volk=from_volk,
            to_volk=to_volk,
            resource=resource,
            reason="Cross-boundary access denied",
            exception_granted=False,
            execution_time_ms=execution_time,
        )

    @traces(
        spec="001",
        story="US-003",
        reqs=["REQ-001-US-003-016", "REQ-001-US-003-017"],
    )
    async def check_permission(
        self, volk: str, resource: str, operation: str
    ) -> PermissionResult:
        """Check permission for operation (REQ-001-US-003-016, REQ-001-US-003-017).

        Args:
            volk: Volk requesting permission
            resource: Resource being accessed
            operation: Operation (read, write, execute)

        Returns:
            PermissionResult with permission status
        """
        start_time = time.time()
        loop = asyncio.get_running_loop()

        # Check permission via Prolog (async)
        query = f"pam:has_permission('{volk}', '{resource}', '{operation}')"
        try:
            has_perm = await loop.run_in_executor(None, self._prolog.query_bool, query)
        except Exception as e:
            logger.error(f"Prolog query failed for permission check: {e}")
            has_perm = False

        execution_time = (time.time() - start_time) * 1000

        if has_perm:
            # Check if temporary grant (async)
            grant_query = f"pam:grant_active('{volk}', '{resource}', ExpiresAt)"
            try:
                grant = await loop.run_in_executor(
                    None, self._prolog.query_one, grant_query
                )
            except Exception as e:
                logger.error(f"Prolog query failed for grant check: {e}")
                grant = None

            if grant:
                expires_at = datetime.fromtimestamp(grant.get("ExpiresAt", 0), tz=UTC)
                return PermissionResult(
                    status=PermissionStatus.TEMPORARY_GRANT,
                    volk=volk,
                    resource=resource,
                    operation=operation,
                    reason=f"Temporary grant until {expires_at}",
                    temporary=True,
                    expires_at=expires_at,
                    execution_time_ms=execution_time,
                )

            return PermissionResult(
                status=PermissionStatus.GRANTED,
                volk=volk,
                resource=resource,
                operation=operation,
                reason=f"Permission granted for {operation}",
                execution_time_ms=execution_time,
            )

        # Log violation (REQ-001-US-003-020)
        await self._log_violation(
            "permission_denied",
            {"volk": volk, "resource": resource, "operation": operation},
        )

        return PermissionResult(
            status=PermissionStatus.DENIED,
            volk=volk,
            resource=resource,
            operation=operation,
            reason=f"Permission denied for {operation}",
            execution_time_ms=execution_time,
        )

    @traces(spec="001", story="US-003", reqs=["REQ-001-US-003-018"])
    async def delegate_ownership(
        self, owner: str, delegate: str, resource: str, duration: int
    ) -> bool:
        """Delegate ownership temporarily (REQ-001-US-003-018).

        Args:
            owner: Current owner
            delegate: Volk receiving delegation
            resource: Resource to delegate
            duration: Duration in seconds

        Returns:
            True if delegation successful
        """
        # Verify owner actually owns resource (must be VALID, not delegated)
        ownership = await self.check_ownership(owner, resource)
        if ownership.status != OwnershipStatus.VALID:
            return False

        # Calculate expiration
        expires_at = (_utc_now() + timedelta(seconds=duration)).timestamp()

        # Assert temporary ownership fact
        from orchestrata_ultimate.core.enums import FactScope

        self._prolog.assert_fact(
            "temporary_ownership",
            [delegate, resource, expires_at, owner],
            FactScope.PERSISTENT,
        )

        logger.info(
            f"Delegated {resource} from {owner} to {delegate} "
            f"for {duration}s (expires: {expires_at})"
        )

        return True

    @traces(spec="001", story="US-003", reqs=["REQ-001-US-003-019"])
    async def grant_temporary_permission(
        self, volk: str, permission: str, duration: int
    ) -> bool:
        """Grant temporary permission (REQ-001-US-003-019).

        Args:
            volk: Volk receiving permission
            permission: Permission to grant (e.g., 'read_write')
            duration: Duration in seconds

        Returns:
            True if grant successful
        """
        # Calculate expiration
        expires_at = (_utc_now() + timedelta(seconds=duration)).timestamp()

        # Assert grant fact
        from orchestrata_ultimate.core.enums import FactScope

        self._prolog.assert_fact(
            "grant_active", [volk, permission, expires_at], FactScope.PERSISTENT
        )

        logger.info(
            f"Granted {permission} to {volk} for {duration}s (expires: {expires_at})"
        )

        return True

    @traces(
        spec="001",
        story="US-003",
        reqs=["REQ-001-US-003-020"],
    )
    async def _log_violation(
        self, violation_type: str, details: dict[str, Any]
    ) -> None:
        """Log permission violation (REQ-001-US-003-020).

        Args:
            violation_type: Type of violation
            details: Violation details
        """
        violation = {
            "type": violation_type,
            "timestamp": _utc_now().isoformat(),
            "details": details,
        }

        self._violations.append(violation)

        # Apply eviction policy
        if len(self._violations) > self._max_violations_before_truncate:
            self._violations = self._violations[-self._violation_log_retain_count :]
            logger.info(
                f"Violation log truncated from {self._max_violations_before_truncate} "
                f"to {self._violation_log_retain_count} entries"
            )

        logger.warning(f"PAM violation: {violation_type} - {details}")

        # Log to graph if available
        if self._graph:
            await self._graph.log_violation(violation)

    @traces(spec="001", story="US-003", reqs=["REQ-001-US-003-020"])
    def get_violations(self) -> list[dict[str, Any]]:
        """Get all logged violations.

        Returns:
            List of violation records
        """
        return self._violations.copy()