orchestrata_ultimate.governance.resource_access

Reference › orchestrata_ultimate › governance › resource_access

Access control for files, network endpoints, APIs and databases, SPEC-007 story US-PGOV-004 (REQ-007-FR4-002, REQ-007-US-004-001). ResourceAccessController implements IResourceAccessController from interfaces and returns an AccessResult from types naming the resource, the mode, the reason and the policy that applied. It asks the Prolog rules in resources.pl through the GovernancePrologClient. No other module imports it yet.

check_file_access, check_network_access, check_api_access and check_db_access query may_access_file/3 and its network, API and database counterparts. For files, when the mode is not a valid bare Prolog atom (checked by _is_valid_prolog_atom) or the Prolog query raises, _python_file_check decides instead, using constants from constants. Its rules in order: a path that starts with a protected prefix is denied; a path containing .. or its encoded form is denied as traversal; reads are allowed; writes are allowed only under an allowed write prefix; anything else is denied by default. The result's policy_applied field says whether Prolog or the Python fallback decided.

At a glance

Class diagram

Dependency diagram

API

Classes

ResourceAccessController

Bases: IResourceAccessController · Decorators: traces(spec='001', story='US-PGOV-004', reqs=['REQ-001-US-003-002']), traces(spec='007', story='US-PGOV-004', reqs=['REQ-007-FR4-002']) · Line 35

Controls access to resources via Prolog policies.

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/governance/resource_access.py

"""Resource access control for Orchestrata governance.

Controls access to files, network, APIs, and databases
via Prolog queries to resources.pl.
"""

import logging
import re

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.governance.constants import (
    DEFAULT_ALLOWED_WRITE_PREFIXES,
    DEFAULT_NETWORK_WHITELIST,
    DEFAULT_PROTECTED_PATHS,
)
from orchestrata_ultimate.governance.interfaces import IResourceAccessController
from orchestrata_ultimate.governance.prolog_client import GovernancePrologClient
from orchestrata_ultimate.governance.types import AccessResult

# Match a valid Prolog atom: lowercase letter followed by word chars.
_PROLOG_ATOM_RE = re.compile(r"^[a-z][a-zA-Z0-9_]*$")


@traces(spec="007", story="US-PGOV-004", reqs=["REQ-007-FR4-002", "REQ-007-US-004-001"])
def _is_valid_prolog_atom(value: str) -> bool:
    """Return True if ``value`` is a well-formed unquoted Prolog atom."""
    return bool(_PROLOG_ATOM_RE.fullmatch(value))


_logger = logging.getLogger(__name__)


@traces(spec="001", story="US-PGOV-004", reqs=["REQ-001-US-003-002"])
@traces(spec="007", story="US-PGOV-004", reqs=["REQ-007-FR4-002"])
class ResourceAccessController(IResourceAccessController):
    """Controls access to resources via Prolog policies.

    Delegates access decisions to resources.pl predicates.
    Falls back to Python-based checks when Prolog is unavailable.
    """

    def __init__(self, prolog_client: GovernancePrologClient) -> None:
        self._prolog_client = prolog_client

    @traces(
        spec="007",
        story="US-PGOV-004",
        reqs=[
            "REQ-007-FR4-002",
            "REQ-007-US-004-001",
        ],
    )
    async def check_file_access(self, actor: str, path: str, mode: str) -> AccessResult:
        """Check if actor can access a file path with given mode."""
        # Reject modes that are not well-formed Prolog atoms. These would
        # either corrupt the Prolog query or produce meaningless results,
        # so fall back to the Python policy instead.
        if not _is_valid_prolog_atom(mode):
            return self._python_file_check(path, mode)

        # Try Prolog first
        try:
            allowed = self._prolog_client.bridge.query_bool(
                f"may_access_file('{actor}', '{path}', {mode})"
            )
            return AccessResult(
                allowed=allowed,
                resource=path,
                mode=mode,
                reason="prolog_policy" if allowed else "prolog_denied",
                policy_applied="resources.pl:may_access_file/3",
            )
        except Exception as exc:
            _logger.warning(
                "Prolog file access check failed, using Python fallback: %s",
                exc,
                exc_info=True,
            )

        # Fallback: Python-based check
        return self._python_file_check(path, mode)

    @traces(
        spec="007",
        story="US-PGOV-004",
        reqs=["REQ-007-US-004-002"],
    )
    async def check_network_access(
        self, actor: str, domain: str, port: int
    ) -> AccessResult:
        """Check if actor can access a network domain:port."""
        try:
            allowed = self._prolog_client.bridge.query_bool(
                f"may_access_network('{actor}', '{domain}', {port})"
            )
            return AccessResult(
                allowed=allowed,
                resource=f"{domain}:{port}",
                mode="connect",
                reason="prolog_policy" if allowed else "prolog_denied",
                policy_applied="resources.pl:may_access_network/3",
            )
        except Exception as exc:
            _logger.warning(
                "Prolog network access check failed, using Python fallback: %s",
                exc,
                exc_info=True,
            )

        # Fallback
        allowed = domain in DEFAULT_NETWORK_WHITELIST
        return AccessResult(
            allowed=allowed,
            resource=f"{domain}:{port}",
            mode="connect",
            reason="whitelisted" if allowed else "not_whitelisted",
            policy_applied="python_fallback",
        )

    @traces(
        spec="007",
        story="US-PGOV-004",
        reqs=["REQ-007-US-004-003"],
    )
    async def check_api_access(
        self, actor: str, endpoint: str, method: str
    ) -> AccessResult:
        """Check if actor can access an API endpoint."""
        try:
            allowed = self._prolog_client.bridge.query_bool(
                f"may_access_api('{actor}', '{endpoint}', {method})"
            )
            return AccessResult(
                allowed=allowed,
                resource=endpoint,
                mode=method,
                reason="prolog_policy" if allowed else "prolog_denied",
                policy_applied="resources.pl:may_access_api/3",
            )
        except Exception as exc:
            _logger.warning(
                "Prolog API access check failed, using Python fallback: %s",
                exc,
                exc_info=True,
            )

        # Fallback: GET always allowed, others need explicit permission
        allowed = method.lower() == "get"
        return AccessResult(
            allowed=allowed,
            resource=endpoint,
            mode=method,
            reason="get_allowed" if allowed else "requires_permission",
            policy_applied="python_fallback",
        )

    @traces(
        spec="007",
        story="US-PGOV-004",
        reqs=["REQ-007-US-004-004"],
    )
    async def check_db_access(
        self, actor: str, schema: str, operation: str
    ) -> AccessResult:
        """Check if actor can perform a database operation."""
        try:
            allowed = self._prolog_client.bridge.query_bool(
                f"may_access_db('{actor}', '{schema}', {operation})"
            )
            return AccessResult(
                allowed=allowed,
                resource=schema,
                mode=operation,
                reason="prolog_policy" if allowed else "prolog_denied",
                policy_applied="resources.pl:may_access_db/3",
            )
        except Exception as exc:
            _logger.warning(
                "Prolog DB access check failed, using Python fallback: %s",
                exc,
                exc_info=True,
            )

        # Fallback: only select and insert allowed
        allowed = operation.lower() in ("select", "insert")
        return AccessResult(
            allowed=allowed,
            resource=schema,
            mode=operation,
            reason="fallback_allowed" if allowed else "fallback_denied",
            policy_applied="python_fallback",
        )

    @traces(spec="007", story="US-PGOV-004", reqs=["REQ-007-US-004-001"])
    def _python_file_check(self, path: str, mode: str) -> AccessResult:
        """Python-based file access check (fallback)."""
        lower_path = path.lower()

        # Check protected paths
        for protected in DEFAULT_PROTECTED_PATHS:
            if lower_path.startswith(protected):
                return AccessResult(
                    allowed=False,
                    resource=path,
                    mode=mode,
                    reason=f"protected_path:{protected}",
                    policy_applied="python_fallback",
                )

        # Check path traversal
        if ".." in path or "%2e%2e" in path.lower():
            return AccessResult(
                allowed=False,
                resource=path,
                mode=mode,
                reason="path_traversal_detected",
                policy_applied="python_fallback",
            )

        # Read is generally allowed
        if mode == "read":
            return AccessResult(
                allowed=True,
                resource=path,
                mode=mode,
                reason="read_allowed",
                policy_applied="python_fallback",
            )

        # Write requires allowed prefix
        if mode == "write":
            for prefix in DEFAULT_ALLOWED_WRITE_PREFIXES:
                if lower_path.startswith(prefix):
                    return AccessResult(
                        allowed=True,
                        resource=path,
                        mode=mode,
                        reason=f"write_prefix:{prefix}",
                        policy_applied="python_fallback",
                    )
            return AccessResult(
                allowed=False,
                resource=path,
                mode=mode,
                reason="no_write_prefix",
                policy_applied="python_fallback",
            )

        # Default deny
        return AccessResult(
            allowed=False,
            resource=path,
            mode=mode,
            reason="default_deny",
            policy_applied="python_fallback",
        )