fix: authenticate the log-stream reconciliation as the service principal

Author: yyyyyyyan

app/api/deps.py

@@ -173,10 +173,9 @@ async def require_admin_for_unsafe_methods(request: Request) -> None:
     :raises HTTPUnauthorizedException: When the method is unsafe and the request
         carries no Bearer credential, or the credential does not validate.
     :raises HTTPForbiddenException: When the resolved user is not an admin, or is
-        inactive. Resolving a user at all is new on routes that previously met
-        only a header check, so both refusals are new there.
+        inactive.
     :raises BaseAuthProviderException: When the auth provider errors while
-        validating the credential, for the same reason.
+        validating the credential.
     """
     if request.method in SAFE_HTTP_METHODS:
         return

app/sep/routes/stream_logs.py

@@ -25,6 +25,7 @@
 from fastapi import APIRouter, Depends, HTTPException, Request
 from starlette.responses import StreamingResponse
 
+from app.core.security import require_internal_token
 from app.sep.deps import (
     ApiCurrentUser,
     get_task_history,
@@ -159,12 +160,20 @@ async def task_history_logs_event_stream(
     Streams log lines for a given task history ID from the Tasks API and yields them
     formatted as server-sent events.
 
+    The log read carries ``access_token`` so it is attributed to the viewing user.
+    The reconciliation that follows it is a mutating Tasks API call, which the
+    unsafe-method admin gate admits only for an admin or the service principal —
+    and this stream is open to any authenticated user, so that call carries the
+    internal token instead.
+
     :param tasks_client: The TaskAPI client for interacting with the Tasks service.
     :type tasks_client: RemoteAPI
     :param task_history_id: The ID of the task history whose logs to stream.
     :type task_history_id: int
     :param request: The FastAPI request object, used to access query parameters.
     :type request: Request
+    :param access_token: Bearer token authenticating the log read as the viewing
+        user.
     :yield: Log entries formatted as server-sent events.
     :rtype: str
     """
@@ -179,7 +188,8 @@ async def task_history_logs_event_stream(
             ):
                 if log_entry:
                     yield f"data: {log_entry.decode()}\n\n"
-            task_history = await tasks_api.post(f"/history/{task_history_id}/sync/")
+        with tasks_client.auth(require_internal_token()) as sync_api:
+            task_history = await sync_api.post(f"/history/{task_history_id}/sync/")
         yield f"event: finish\ndata: {json.dumps({'status': task_history['status']})}\n\n"
     except TimeoutError as exc:
         logger.warning(

tests/app/api/test_admin_gate.py

@@ -100,7 +100,7 @@ async def test_exactly_one_route_is_exempt_from_the_gate() -> None:
     raises for every other, which also pins that ``allow_non_admin_mutation``
     registered the object FastAPI stores as ``APIRoute.endpoint``.
     """
-    exempt = set()
+    exempt: set[Callable[..., Any]] = set()
     for app in (sep_app, inventory_app, tasks_app):
         for route in _api_routes(app):
             request = make_request("POST", endpoint=route.endpoint)

tests/app/sep/routes/test_stream_logs.py

@@ -102,6 +102,49 @@ def test_archives_logs_event_stream(
     )
 
 
+def test_sync_hop_is_authenticated_as_the_service_principal(
+    test_client, mock_tasks_client, task_history_response, mocker
+):
+    """Send the end-of-stream reconciliation under the internal token.
+
+    The log stream is reachable by any authenticated user, and the ``sync`` hop
+    it issues at stream end is a mutating request on the Tasks API — which is
+    admin-gated. Carrying the viewing user's own bearer there ends a non-admin's
+    stream in an error frame instead of the finish frame, so the hop is
+    authenticated as the service principal the gate admits by identity.
+
+    Evidence stops at the credential the request carries; the gate's own
+    treatment of that identity is covered in ``tests/app/tasks/test_admin_gate.py``.
+    """
+    mocker.patch(
+        "app.sep.routes.stream_logs.require_internal_token",
+        return_value="internal-token",
+    )
+    active_tokens = []
+    sync_token = {}
+
+    @contextmanager
+    def recording_auth(token: str):
+        active_tokens.append(token)
+        try:
+            yield mock_tasks_client
+        finally:
+            active_tokens.pop()
+
+    async def recording_post(_path, **_kwargs):
+        sync_token["value"] = active_tokens[-1]
+        return task_history_response.model_dump()
+
+    mock_tasks_client.auth = recording_auth
+    mock_tasks_client.post.side_effect = recording_post
+
+    response = test_client.get(f"/stream-logs/{task_history_response.id}")
+
+    assert response.status_code == HTTP_200_OK
+    assert "event: finish" in response.text
+    assert sync_token["value"] == "internal-token"
+
+
 def test_logs_event_stream_emits_sep_error_on_upstream_error(
     test_client, mock_tasks_client, task_history_response
 ):

tests/app/tasks/test_admin_gate.py

@@ -166,6 +166,45 @@ async def fake_dispatch_queue_item(queue_item, passed_session):
     assert response.json()["status"] == TaskHistoryStatusEnum.RUNNING.value
 
 
+def test_the_log_stream_reconciliation_is_refused_for_a_non_admin(
+    bearer_client: TestClient, created_task_with_history
+) -> None:
+    """Refuse the task-history reconciliation for a non-admin.
+
+    It is a genuine write — it persists ``status``, ``started_at`` and
+    ``finished_at`` — so it stays gated rather than joining the exemption
+    allowlist. The SEP log stream that triggers it is open to any authenticated
+    user, which is why that caller sends the internal token instead.
+    """
+    response = bearer_client.post(
+        f"/history/{created_task_with_history.id}/sync/", headers=BEARER_HEADERS
+    )
+
+    assert response.status_code == status.HTTP_403_FORBIDDEN
+
+
+def test_the_log_stream_reconciliation_is_accepted_for_the_service_principal(
+    bearer_client: TestClient,
+    created_task_with_history,
+    mock_executor: AsyncMock,
+    mocker: MockerFixture,
+) -> None:
+    """Accept the same reconciliation when it carries the internal token.
+
+    This is the identity the SEP log stream sends, so a non-admin's stream still
+    reaches its finish frame.
+    """
+    mocker.patch.object(settings, "SEP_INTERNAL_TOKEN", SecretStr(SERVICE_TOKEN))
+    mock_executor.sync_task_history.return_value = created_task_with_history
+
+    response = bearer_client.post(
+        f"/history/{created_task_with_history.id}/sync/",
+        headers={"Authorization": f"Bearer {SERVICE_TOKEN}"},
+    )
+
+    assert response.status_code == status.HTTP_200_OK
+
+
 def test_the_batch_read_stays_reachable_for_a_non_admin(
     bearer_client: TestClient,
 ) -> None: