diff --git a/macfleet/compute/authz.py b/macfleet/compute/authz.py index 62d9ee1..f0aeb0f 100644 --- a/macfleet/compute/authz.py +++ b/macfleet/compute/authz.py @@ -3,6 +3,7 @@ from __future__ import annotations import math +from collections.abc import Iterable from dataclasses import dataclass, field from typing import Optional @@ -11,6 +12,24 @@ class TaskAuthorizationError(PermissionError): """Raised when a worker refuses to execute a task.""" +def _normalize_task_names( + field_name: str, + value: Optional[Iterable[str]], + *, + allow_none: bool, +) -> Optional[frozenset[str]]: + if value is None: + if allow_none: + return None + return frozenset() + if isinstance(value, (str, bytes)) or not isinstance(value, Iterable): + raise ValueError(f"{field_name} must be a collection of task names") + normalized = frozenset(value) + if any(not isinstance(task_name, str) or not task_name for task_name in normalized): + raise ValueError(f"{field_name} must contain only non-empty task names") + return normalized + + @dataclass(frozen=True) class TaskAuthorizationPolicy: """Worker-side authorization policy for registered tasks. @@ -27,14 +46,25 @@ class TaskAuthorizationPolicy: def __post_init__(self) -> None: if not isinstance(self.role, str) or not self.role: raise ValueError("role must be a non-empty string") - for name, task_names in ( - ("allowed_tasks", self.allowed_tasks), - ("denied_tasks", self.denied_tasks), - ): - if task_names is not None and any( - not isinstance(task_name, str) or not task_name for task_name in task_names - ): - raise ValueError(f"{name} must contain only non-empty task names") + allowed_tasks = _normalize_task_names( + "allowed_tasks", + self.allowed_tasks, + allow_none=True, + ) + denied_tasks = _normalize_task_names( + "denied_tasks", + self.denied_tasks, + allow_none=False, + ) + assert denied_tasks is not None + overlap = sorted((allowed_tasks or frozenset()) & denied_tasks) + if overlap: + raise ValueError( + "allowed_tasks and denied_tasks overlap: " + f"{', '.join(overlap)}" + ) + object.__setattr__(self, "allowed_tasks", allowed_tasks) + object.__setattr__(self, "denied_tasks", denied_tasks) if ( isinstance(self.max_timeout_sec, bool) or not isinstance(self.max_timeout_sec, (int, float)) diff --git a/macfleet/compute/worker.py b/macfleet/compute/worker.py index a938614..14919f7 100644 --- a/macfleet/compute/worker.py +++ b/macfleet/compute/worker.py @@ -24,7 +24,7 @@ from macfleet.comm.protocol import MessageType from macfleet.comm.transport import PeerTransport -from macfleet.compute.authz import DEFAULT_TASK_POLICY, TaskAuthorizationPolicy +from macfleet.compute.authz import DEFAULT_TASK_POLICY, TaskAuthorizationError, TaskAuthorizationPolicy from macfleet.compute.models import TaskNotRegisteredError, TaskResult, TaskSpec from macfleet.security.audit import audit_event @@ -190,6 +190,14 @@ async def _execute_and_reply(self, spec: TaskSpec) -> None: ok=False, error=f"Task timed out after {spec.timeout_sec}s", ) + except TaskAuthorizationError as e: + audit_event( + "task.authorization_denied", + task_name=spec.task_name, + coordinator=self._coordinator, + reason=str(e), + ) + result = TaskResult.failure(spec.task_id, e) except Exception as e: audit_event( "task.failed", diff --git a/tests/test_compute/test_authz.py b/tests/test_compute/test_authz.py index 108353b..aaeeb1a 100644 --- a/tests/test_compute/test_authz.py +++ b/tests/test_compute/test_authz.py @@ -12,7 +12,17 @@ [ ({"role": ""}, "role"), ({"allowed_tasks": frozenset({""})}, "allowed_tasks"), + ({"allowed_tasks": "task.name"}, "allowed_tasks"), + ({"allowed_tasks": 42}, "allowed_tasks"), ({"denied_tasks": frozenset({""})}, "denied_tasks"), + ({"denied_tasks": "task.name"}, "denied_tasks"), + ( + { + "allowed_tasks": frozenset({"task.a"}), + "denied_tasks": frozenset({"task.a"}), + }, + "overlap", + ), ({"max_timeout_sec": 0}, "max_timeout_sec"), ({"max_timeout_sec": float("nan")}, "max_timeout_sec"), ], @@ -20,3 +30,13 @@ def test_policy_rejects_invalid_configuration(kwargs, message): with pytest.raises(ValueError, match=message): TaskAuthorizationPolicy(**kwargs) + + +def test_policy_normalizes_task_name_collections(): + policy = TaskAuthorizationPolicy( + allowed_tasks=["task.a", "task.b"], + denied_tasks=("task.c",), + ) + + assert policy.allowed_tasks == frozenset({"task.a", "task.b"}) + assert policy.denied_tasks == frozenset({"task.c"}) diff --git a/tests/test_compute/test_dispatch.py b/tests/test_compute/test_dispatch.py index c188249..38867b5 100644 --- a/tests/test_compute/test_dispatch.py +++ b/tests/test_compute/test_dispatch.py @@ -175,8 +175,13 @@ async def test_remote_disabled_task_is_refused(self): await _teardown(coordinator, worker) @pytest.mark.asyncio - async def test_worker_allowlist_blocks_unlisted_task(self): + async def test_worker_allowlist_blocks_unlisted_task(self, monkeypatch): """Workers can restrict execution to an explicit task allowlist.""" + audit_events = [] + monkeypatch.setattr( + "macfleet.compute.worker.audit_event", + lambda event, **fields: audit_events.append((event, fields)), + ) coordinator, worker, _ = await _setup_pair() try: dispatcher = TaskDispatcher(coordinator, ["worker-0"]) @@ -195,6 +200,14 @@ async def test_worker_allowlist_blocks_unlisted_task(self): with pytest.raises(RemoteTaskError) as exc_info: await blocked.result(timeout=5.0) assert "not in the worker allowlist" in exc_info.value.remote_traceback + assert ( + "task.authorization_denied", + { + "task_name": pow_two.task_name, + "coordinator": "coordinator", + "reason": f"Task {pow_two.task_name!r} is not in the worker allowlist", + }, + ) in audit_events await tw.stop() await dispatcher.stop()