Add per-resource access control primitives and run scopes

cloud_pipelines_backend/access_control/checks.py

@@ -0,0 +1,79 @@
+import dataclasses
+from typing import Any
+
+from .. import errors
+from . import keys
+from . import scopes
+
+
+class NotAuthorizedError(errors.PermissionError):
+    """A failed scope check; `owner` is carried for the host to show or withhold."""
+
+    def __init__(self, message: str, *, owner: str | None = None) -> None:
+        super().__init__(message)
+        self.owner = owner
+
+
+@dataclasses.dataclass(frozen=True, kw_only=True)
+class Caller:
+    """Who is asking; is_admin holds every scope, for hosts that pass it rather than skip the check."""
+
+    email: str | None
+    is_admin: bool = False
+
+
+def normalize_email(*, email: str) -> str:
+    """Lowercases and trims an email so the same address always compares equal."""
+    return email.strip().lower()
+
+
+def is_owner(*, owner: str | None, caller: Caller) -> bool:
+    """True when the caller's email matches the owner's, ignoring case and whitespace."""
+    return bool(
+        owner
+        and caller.email
+        and normalize_email(email=owner) == normalize_email(email=caller.email)
+    )
+
+
+def effective_scopes(
+    *,
+    resource_type: scopes.ResourceType,
+    owner: str | None,
+    access_control: dict[str, Any] | None,
+    caller: Caller,
+) -> frozenset[str]:
+    """Fine scopes the caller holds: all for the owner and admins, else what their grant expands to."""
+    if caller.is_admin or is_owner(owner=owner, caller=caller):
+        return resource_type.fine_scopes
+    if not caller.email:
+        return frozenset()
+    users = (access_control or {}).get(keys.Key.USERS) or {}
+    entry = users.get(normalize_email(email=caller.email)) or {}
+    granted: set[str] = set()
+    for scope in entry.get(keys.Key.PERMISSIONS) or []:
+        granted |= resource_type.expand(scope=scope)
+    return frozenset(granted)
+
+
+def require(
+    *,
+    scope: str,
+    resource_type: scopes.ResourceType,
+    resource_id: str,
+    owner: str | None,
+    access_control: dict[str, Any] | None,
+    caller: Caller,
+) -> None:
+    """Raises NotAuthorizedError unless the caller holds the scope on the resource."""
+    granted = effective_scopes(
+        resource_type=resource_type,
+        owner=owner,
+        access_control=access_control,
+        caller=caller,
+    )
+    if scope not in granted:
+        raise NotAuthorizedError(
+            f"{caller.email} does not have {scope} on {resource_type.name} {resource_id}.",
+            owner=owner,
+        )

cloud_pipelines_backend/access_control/edits.py

@@ -0,0 +1,98 @@
+import datetime
+from typing import Any, Final
+
+from .. import errors
+from . import checks
+from . import keys
+from . import scopes
+
+MAX_HISTORY_ENTRIES: Final[int] = 100
+
+
+class StaleRevisionError(Exception):
+    pass
+
+
+def latest_revision(*, access_control: dict[str, Any] | None) -> int:
+    """Revision of the newest history entry, which is always last, or 0 when access has never been changed."""
+    history = (access_control or {}).get(keys.Key.HISTORY) or []
+    return history[-1][keys.Key.REVISION] if history else 0
+
+
+def _validated_users(
+    *,
+    resource_type: scopes.ResourceType,
+    owner: str | None,
+    users: dict[str, dict[str, list[str]]],
+) -> dict[str, dict[str, list[str]]]:
+    """Normalizes a requested user list and drops empty grants.
+
+    Rejects blank emails, the owner, duplicates and unknown scopes.
+    """
+    validated: dict[str, dict[str, list[str]]] = {}
+    for raw_email, entry in users.items():
+        email = checks.normalize_email(email=raw_email)
+        if not email:
+            raise errors.ApiValidationError("User email must not be empty.")
+        if owner and email == checks.normalize_email(email=owner):
+            raise errors.ApiValidationError(
+                f"{email} owns this {resource_type.name} and cannot be listed."
+            )
+        if email in validated:
+            raise errors.ApiValidationError(f"{email} is listed more than once.")
+        permissions = sorted(set(entry.get(keys.Key.PERMISSIONS) or []))
+        unknown = [
+            scope for scope in permissions if not resource_type.is_known(scope=scope)
+        ]
+        if unknown:
+            raise errors.ApiValidationError(
+                f"Unknown permissions for {resource_type.name}: {', '.join(unknown)}."
+            )
+        if permissions:
+            validated[email] = {keys.Key.PERMISSIONS: permissions}
+    return validated
+
+
+def apply_users_edit(
+    *,
+    resource_type: scopes.ResourceType,
+    owner: str | None,
+    access_control: dict[str, Any] | None,
+    users: dict[str, dict[str, list[str]]],
+    changed_by: str,
+    changed_at: datetime.datetime,
+    expected_revision: int,
+) -> dict[str, Any]:
+    """Replaces the user list at expected_revision and logs each user's from/to as one history entry.
+
+    The revision check only sees this snapshot, so the caller must hold a row lock across the read
+    that produced access_control and the write of the result. Keeps the newest MAX_HISTORY_ENTRIES.
+    """
+    current = access_control or {}
+    revision = latest_revision(access_control=current)
+    if expected_revision != revision:
+        raise StaleRevisionError(
+            f"Access was changed since revision {expected_revision}; the latest is {revision}."
+        )
+    old_users = current.get(keys.Key.USERS) or {}
+    new_users = _validated_users(resource_type=resource_type, owner=owner, users=users)
+    changes = {}
+    for email in sorted(old_users.keys() | new_users.keys()):
+        before = (old_users.get(email) or {}).get(keys.Key.PERMISSIONS) or []
+        after = (new_users.get(email) or {}).get(keys.Key.PERMISSIONS) or []
+        if before != after:
+            changes[email] = {keys.Key.FROM: before, keys.Key.TO: after}
+    history = list(current.get(keys.Key.HISTORY) or [])
+    if changes:
+        history.append(
+            {
+                keys.Key.REVISION: revision + 1,
+                keys.Key.CHANGED_BY: checks.normalize_email(email=changed_by),
+                keys.Key.CHANGED_AT: changed_at.isoformat(),
+                keys.Key.CHANGES: changes,
+            }
+        )
+    return {
+        keys.Key.USERS: new_users,
+        keys.Key.HISTORY: history[-MAX_HISTORY_ENTRIES:],
+    }

cloud_pipelines_backend/access_control/keys.py

@@ -0,0 +1,16 @@
+import enum
+
+
+class Key(str, enum.Enum):
+    __hash__ = str.__hash__
+    __str__ = str.__str__
+
+    USERS = "users"
+    PERMISSIONS = "permissions"
+    HISTORY = "history"
+    REVISION = "revision"
+    CHANGED_BY = "changed_by"
+    CHANGED_AT = "changed_at"
+    CHANGES = "changes"
+    FROM = "from"
+    TO = "to"

cloud_pipelines_backend/access_control/runs.py

@@ -0,0 +1,120 @@
+import dataclasses
+import enum
+from collections import abc
+from typing import Any, Protocol
+
+from sqlalchemy import orm
+
+from .. import backend_types_sql as bts
+from . import checks
+from . import scopes
+
+
+class RunScope(str, enum.Enum):
+    __hash__ = str.__hash__
+    __str__ = str.__str__
+
+    CANCEL = "run:cancel"
+    ANNOTATE = "run:annotate"
+
+
+@dataclasses.dataclass(frozen=True, kw_only=True)
+class RunParent:
+    resource_type: scopes.ResourceType
+    owner: str | None
+    access_control: dict[str, Any] | None
+
+
+class RunParentResolver(Protocol):
+    def __call__(
+        self, *, session: orm.Session, pipeline_run: bts.PipelineRun
+    ) -> abc.Iterable[RunParent]: ...
+
+
+_parent_resolvers: list[RunParentResolver] = []
+_parent_annotation_keys: set[str] = set()
+
+
+def register_run_parent_resolver(
+    *, resolver: RunParentResolver, annotation_keys: abc.Iterable[str] = ()
+) -> None:
+    """Adds a host hook that finds the resources a run inherits run:* scopes from, via the given annotations."""
+    if resolver not in _parent_resolvers:
+        _parent_resolvers.append(resolver)
+    _parent_annotation_keys.update(annotation_keys)
+
+
+def _is_starter_or_admin(
+    *, pipeline_run: bts.PipelineRun, caller: checks.Caller
+) -> bool:
+    """True for admins and whoever started the run."""
+    return (
+        caller.is_admin
+        or caller.email == pipeline_run.created_by
+        or checks.is_owner(owner=pipeline_run.created_by, caller=caller)
+    )
+
+
+def run_scopes(
+    *,
+    session: orm.Session,
+    pipeline_run: bts.PipelineRun,
+    caller: checks.Caller,
+) -> frozenset[str]:
+    """Run scopes the caller holds: all for the starter and admins, else those inherited from parents."""
+    all_scopes = frozenset(scope.value for scope in RunScope)
+    if _is_starter_or_admin(pipeline_run=pipeline_run, caller=caller):
+        return all_scopes
+    granted: set[str] = set()
+    for resolver in _parent_resolvers:
+        for parent in resolver(session=session, pipeline_run=pipeline_run):
+            granted |= checks.effective_scopes(
+                resource_type=parent.resource_type,
+                owner=parent.owner,
+                access_control=parent.access_control,
+                caller=caller,
+            )
+    return frozenset(granted) & all_scopes
+
+
+def require_run_scope(
+    *,
+    session: orm.Session,
+    pipeline_run: bts.PipelineRun,
+    caller: checks.Caller,
+    scope: RunScope,
+) -> None:
+    """Raises NotAuthorizedError unless the caller holds the scope on the run."""
+    if scope not in run_scopes(
+        session=session, pipeline_run=pipeline_run, caller=caller
+    ):
+        raise checks.NotAuthorizedError(
+            f"The pipeline run {pipeline_run.id} was started by {pipeline_run.created_by} and {caller.email} does not have {scope} on it.",
+            owner=pipeline_run.created_by,
+        )
+
+
+def require_annotation_change(
+    *,
+    session: orm.Session,
+    pipeline_run: bts.PipelineRun,
+    caller: checks.Caller,
+    key: str,
+) -> None:
+    """Raises NotAuthorizedError unless the caller may set or delete the annotation.
+
+    Annotations a parent resolver reads are kept to the starter and admins, so a grantee can't re-parent the run.
+    """
+    require_run_scope(
+        session=session,
+        pipeline_run=pipeline_run,
+        caller=caller,
+        scope=RunScope.ANNOTATE,
+    )
+    if key in _parent_annotation_keys and not _is_starter_or_admin(
+        pipeline_run=pipeline_run, caller=caller
+    ):
+        raise checks.NotAuthorizedError(
+            f"Only {pipeline_run.created_by} can change the {key} annotation on pipeline run {pipeline_run.id}.",
+            owner=pipeline_run.created_by,
+        )

cloud_pipelines_backend/access_control/scopes.py

@@ -0,0 +1,69 @@
+import dataclasses
+import enum
+
+from .. import errors
+
+
+class BroadScope(str, enum.Enum):
+    __hash__ = str.__hash__
+    __str__ = str.__str__
+
+    OPERATE = "operate"
+    EDIT = "edit"
+    MANAGE = "manage"
+
+
+@dataclasses.dataclass(frozen=True, kw_only=True)
+class ResourceType:
+    name: str
+    operate: frozenset[str]
+    edit: frozenset[str]
+    manage: frozenset[str]
+
+    @property
+    def fine_scopes(self) -> frozenset[str]:
+        """Every fine scope this resource type defines."""
+        return self.operate | self.edit | self.manage
+
+    def expand(self, *, scope: str) -> frozenset[str]:
+        """Fine scopes a grant stands for; a broad scope includes lower tiers, an unknown one nothing."""
+        match scope:
+            case BroadScope.OPERATE:
+                return self.operate
+            case BroadScope.EDIT:
+                return self.operate | self.edit
+            case BroadScope.MANAGE:
+                return self.fine_scopes
+        return frozenset({scope}) & self.fine_scopes
+
+    def is_known(self, *, scope: str) -> bool:
+        """True for a broad scope or one of this type's fine scopes."""
+        return scope in set(BroadScope) or scope in self.fine_scopes
+
+
+_resource_types: dict[str, ResourceType] = {}
+
+
+def register_resource_type(
+    *,
+    name: str,
+    operate: frozenset[str],
+    edit: frozenset[str],
+    manage: frozenset[str],
+) -> ResourceType:
+    """Declares a resource type's fine scopes per tier; re-declaring it with different scopes raises."""
+    resource_type = ResourceType(name=name, operate=operate, edit=edit, manage=manage)
+    existing = _resource_types.get(name)
+    if existing is not None and existing != resource_type:
+        raise errors.ApiValidationError(
+            f"Resource type {name} is already registered with different scopes."
+        )
+    _resource_types[name] = resource_type
+    return resource_type
+
+
+def get_resource_type(*, name: str) -> ResourceType:
+    """Looks up a registered resource type, raising ItemNotFoundError if none has that name."""
+    if name not in _resource_types:
+        raise errors.ItemNotFoundError(f"Unknown resource type {name}.")
+    return _resource_types[name]