""" Session scoping for experimental task identifiers. Task IDs generated by `run_task()` embed an opaque, per-session marker (the "session scope") so that the default task handlers can tell which session created a task. The default handlers for tasks/get, tasks/result, tasks/list, and tasks/cancel only operate on tasks created by the requesting session. Task IDs without a session scope (explicitly provided IDs, IDs created directly through a TaskStore, or IDs created in stateless mode) have no known creator. They can be used with tasks/get, tasks/result, and tasks/cancel from any session - possession of the ID is what grants access - but they are never included in tasks/list responses. WARNING: These APIs are experimental and may change without notice. """ import re from uuid import uuid4 __all__ = [ "new_session_scope", "scoped_task_id", "session_scope_of", "task_in_session_scope", "task_listable_in_session_scope", ] # A scoped task ID has the form "<32 hex chars>:". Both halves must # match exactly so that explicitly chosen task IDs are never mistaken for # scoped ones. \Z rather than $ so a trailing newline cannot match. _SCOPED_TASK_ID = re.compile( r"\A(?P[0-9a-f]{32}):" r"[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}\Z" ) def new_session_scope() -> str: """Create a new opaque session scope token.""" return uuid4().hex def scoped_task_id(session_scope: str) -> str: """Generate a task ID associated with the given session scope.""" return f"{session_scope}:{uuid4()}" def session_scope_of(task_id: str) -> str | None: """Return the session scope embedded in a task ID, or None if it has none.""" match = _SCOPED_TASK_ID.match(task_id) return match.group("scope") if match else None def task_in_session_scope(task_id: str, session_scope: str | None) -> bool: """Whether a task may be used by a requestor with the given session scope. Used by tasks/get, tasks/result, and tasks/cancel. A task whose ID carries no session scope has no known creator, so possession of the ID is what grants access to it: it can be used from any session. """ embedded = session_scope_of(task_id) return embedded is None or embedded == session_scope def task_listable_in_session_scope(task_id: str, session_scope: str | None) -> bool: """Whether a task may be included in a tasks/list response for the given session scope. Used by tasks/list. Listing is stricter than access by ID: a task is only listed to the session that created it. Tasks with no session scope are never listed because they have no known creator, and requestors with no session scope are never shown any tasks because the server cannot tell them apart. """ embedded = session_scope_of(task_id) return embedded is not None and embedded == session_scope