aios/active-runtime/app/task_plane/scope.py

"""Trusted TaskPlane scope model and SQL authorization helpers."""

from __future__ import annotations

from dataclasses import dataclass, replace
from typing import Any, Mapping

from .errors import ForbiddenError, ValidationError

SCOPE_FIELDS = ("tenant_id", "user_id", "project_id", "conversation_id", "run_id")
PRIVILEGED_CAPABILITIES = frozenset(
    {"*", "admin", "system.admin", "task.admin", "task.read.all"}
)


def _value(value: Any, field: str) -> Any:
    if hasattr(value, "task_plan"):
        value = value.task_plan
    if isinstance(value, Mapping):
        return value.get(field)
    return getattr(value, field, None)


def _text(value: Any) -> str | None:
    if value is None:
        return None
    result = str(value).strip()
    return result or None


@dataclass(frozen=True)
class TaskScope:
    scope_type: str
    tenant_id: str | None = None
    user_id: str | None = None
    project_id: str | None = None
    conversation_id: str | None = None
    run_id: str | None = None


@dataclass(frozen=True)
class TaskActorContext:
    identity: str
    scope_type: str = "legacy"
    tenant_id: str | None = None
    user_id: str | None = None
    project_id: str | None = None
    conversation_id: str | None = None
    run_id: str | None = None
    capabilities: frozenset[str] = frozenset()


def normalize_task_scope(value: Any) -> TaskScope:
    fields = {field: _text(_value(value, field)) for field in SCOPE_FIELDS}
    declared = _text(_value(value, "scope_type"))
    if declared is None:
        declared = "tenant" if fields["tenant_id"] else "legacy"
    if declared == "tenant":
        if fields["tenant_id"] is None:
            raise ValidationError(
                "tenant scope requires tenant_id", code="TASK_SCOPE_TENANT_ID_REQUIRED"
            )
    elif declared == "legacy":
        if any(fields.values()):
            raise ValidationError(
                "legacy scope cannot contain scoped ownership",
                code="TASK_SCOPE_LEGACY_WITH_SCOPED_FIELDS",
            )
    elif declared == "system":
        if any(fields[field] for field in ("tenant_id", "user_id", "project_id", "conversation_id")):
            raise ValidationError(
                "system scope cannot contain tenant ownership",
                code="TASK_SCOPE_SYSTEM_WITH_TENANT_FIELDS",
            )
    else:
        raise ValidationError("invalid task scope type", code="TASK_SCOPE_TYPE_INVALID")
    return TaskScope(declared, **fields)


def actor_is_privileged(actor: TaskActorContext) -> bool:
    return bool(actor.capabilities & PRIVILEGED_CAPABILITIES)


def actor_context(
    identity: str,
    configured: TaskActorContext | Mapping[str, Any] | None,
    capabilities: set[str] | frozenset[str],
) -> TaskActorContext:
    if isinstance(configured, TaskActorContext):
        context = configured
    else:
        data = dict(configured or {})
        data.pop("identity", None)
        data.pop("capabilities", None)
        context = TaskActorContext(identity=str(identity), **data)
    scope = normalize_task_scope(context)
    return replace(
        context,
        identity=str(identity),
        scope_type=scope.scope_type,
        tenant_id=scope.tenant_id,
        user_id=scope.user_id,
        project_id=scope.project_id,
        conversation_id=scope.conversation_id,
        run_id=scope.run_id,
        capabilities=frozenset(str(item) for item in capabilities),
    )


def bind_task_scope(plan: Any, actor: TaskActorContext, *, public_create: bool) -> Any:
    requested = normalize_task_scope(plan)
    if actor_is_privileged(actor):
        trusted = normalize_task_scope(actor)
        if requested.scope_type == "legacy" and trusted.scope_type != "legacy":
            plan.scope_type = trusted.scope_type
            for field in SCOPE_FIELDS:
                setattr(plan, field, getattr(actor, field))
        return plan
    trusted = normalize_task_scope(actor)
    if public_create and trusted.scope_type == "legacy":
        raise ForbiddenError(
            "authenticated identity has no tenant scope",
            code="TASK_ACTOR_SCOPE_UNAVAILABLE",
        )
    if trusted.scope_type != "tenant":
        if requested.scope_type == "legacy":
            plan.scope_type = trusted.scope_type
            for field in SCOPE_FIELDS:
                setattr(plan, field, getattr(actor, field))
            return plan
        if requested.scope_type != trusted.scope_type:
            raise ForbiddenError("task scope mismatch", code="TASK_SCOPE_MISMATCH_DENY")
        return plan
    if requested.scope_type not in {"legacy", "tenant"}:
        raise ForbiddenError("task scope mismatch", code="TASK_SCOPE_MISMATCH_DENY")
    updates: dict[str, Any] = {"scope_type": "tenant"}
    for field in SCOPE_FIELDS:
        declared = getattr(requested, field)
        authoritative = getattr(actor, field)
        if declared is not None and declared != authoritative:
            raise ForbiddenError(
                "task scope mismatch",
                code="TASK_SCOPE_MISMATCH_DENY",
                details={"field": field},
            )
        updates[field] = authoritative
    for field, item in updates.items():
        setattr(plan, field, item)
    return plan


def actor_scope_sql(actor: TaskActorContext | None) -> tuple[str, tuple[Any, ...]]:
    if actor is None or actor_is_privileged(actor):
        return "", ()
    scope = normalize_task_scope(actor)
    if scope.scope_type == "legacy":
        return "scope_type = 'legacy'", ()
    if scope.scope_type == "system":
        return "scope_type = 'system'", ()
    clauses = ["scope_type = 'tenant'", "tenant_id = ?"]
    params: list[Any] = [scope.tenant_id]
    for field in ("user_id", "project_id", "conversation_id", "run_id"):
        value = getattr(scope, field)
        if value is None:
            clauses.append(f"{field} IS NULL")
        else:
            clauses.append(f"({field} IS NULL OR {field} = ?)")
            params.append(value)
    return " AND ".join(clauses), tuple(params)


def task_scope_values(value: Any) -> tuple[str | None, ...]:
    scope = normalize_task_scope(value)
    return (
        scope.scope_type,
        scope.tenant_id,
        scope.user_id,
        scope.project_id,
        scope.conversation_id,
        scope.run_id,
    )