Tighten task authorization policy validation

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))

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",

tests/test_compute/test_authz.py

@@ -12,11 +12,31 @@
     [
         ({"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"),
     ],
 )
 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"})

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()