fix(awooop): authenticate approval decisions

apps/api/tests/test_awooop_operator_auth.py

@@ -0,0 +1,102 @@
+from __future__ import annotations
+
+import pytest
+from fastapi import HTTPException
+
+from src.api.v1.platform import operator_runs
+from src.core.awooop_operator_auth import (
+    AwoooPOperatorPrincipal,
+    authenticate_awooop_operator_headers,
+)
+
+
+def test_operator_auth_rejects_missing_key_in_prod() -> None:
+    with pytest.raises(HTTPException) as exc:
+        authenticate_awooop_operator_headers(
+            operator_id="ops@example.com",
+            operator_key=None,
+            configured_key="",
+            environment="prod",
+        )
+
+    assert exc.value.status_code == 401
+
+
+def test_operator_auth_rejects_invalid_key() -> None:
+    with pytest.raises(HTTPException) as exc:
+        authenticate_awooop_operator_headers(
+            operator_id="ops@example.com",
+            operator_key="wrong",
+            configured_key="expected",
+            environment="prod",
+        )
+
+    assert exc.value.status_code == 401
+
+
+def test_operator_auth_returns_principal_for_valid_key() -> None:
+    principal = authenticate_awooop_operator_headers(
+        operator_id="telegram:123456",
+        operator_key="expected",
+        configured_key="expected",
+        environment="prod",
+    )
+
+    assert principal == AwoooPOperatorPrincipal(
+        operator_id="telegram:123456",
+        auth_method="operator_api_key",
+    )
+
+
+def test_operator_auth_rejects_malformed_identity() -> None:
+    with pytest.raises(HTTPException) as exc:
+        authenticate_awooop_operator_headers(
+            operator_id="../operator",
+            operator_key="expected",
+            configured_key="expected",
+            environment="prod",
+        )
+
+    assert exc.value.status_code == 422
+
+
+@pytest.mark.asyncio
+async def test_decide_approval_uses_trusted_principal_not_body(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    captured: dict[str, object] = {}
+
+    async def fake_decide_approval_svc(**kwargs: object) -> dict[str, object]:
+        captured.update(kwargs)
+        return {
+            "run_id": kwargs["run_id"],
+            "decision": kwargs["decision"],
+            "new_state": "running",
+            "approval_token_jti": "jti-test",
+        }
+
+    monkeypatch.setattr(
+        operator_runs,
+        "decide_approval_svc",
+        fake_decide_approval_svc,
+    )
+
+    body = operator_runs.DecideApprovalRequest(
+        project_id="awoooi",
+        decision="approve",
+        approver_id="spoofed-browser-value",
+    )
+    principal = AwoooPOperatorPrincipal(
+        operator_id="telegram:123456",
+        auth_method="operator_api_key",
+    )
+
+    response = await operator_runs.decide_approval(
+        "018f2d04-4c37-7a18-b764-df0df0cbe111",
+        body,
+        principal,
+    )
+
+    assert response["approval_token_jti"] == "jti-test"
+    assert captured["approver_id"] == "telegram:123456"
+    assert captured["project_id"] == "awoooi"