activity-core/tests/test_ops_runs_api.py

"""Authentication boundary tests for the ops_run worker API."""

from __future__ import annotations

from unittest.mock import MagicMock

import pytest
from fastapi import FastAPI, HTTPException
from fastapi.testclient import TestClient

from activity_core.ops_run_queue import CloseOpsRunOutcome
from activity_core.ops_runs_api import (
    _close_response,
    bind_worker_id,
    require_worker,
    require_worker_or_operator,
    router,
)


def _request(headers: dict[str, str] | None = None) -> MagicMock:
    request = MagicMock()
    request.headers = headers or {}
    return request


def _client() -> TestClient:
    app = FastAPI()
    app.include_router(router)
    return TestClient(app)


def test_worker_token_binds_configured_identity(monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setenv("ACTIVITY_CORE_WORKER_TOKEN", "worker-secret")
    monkeypatch.setenv("ACTIVITY_CORE_WORKER_ID", "rein-aharness@railiance01")

    authenticated = require_worker(
        _request(),
        x_worker_token="worker-secret",
    )

    assert authenticated == "rein-aharness@railiance01"
    assert bind_worker_id("rein-aharness@railiance01", authenticated) == authenticated


def test_spoofed_worker_id_is_rejected(monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setenv("ACTIVITY_CORE_WORKER_TOKEN", "worker-secret")
    monkeypatch.setenv("ACTIVITY_CORE_WORKER_ID", "rein-aharness@railiance01")
    authenticated = require_worker(_request(), x_worker_token="worker-secret")

    with pytest.raises(HTTPException) as exc:
        bind_worker_id("another-worker", authenticated)

    assert exc.value.status_code == 403


def test_worker_token_without_identity_fails_closed(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    monkeypatch.setenv("ACTIVITY_CORE_WORKER_TOKEN", "worker-secret")
    monkeypatch.delenv("ACTIVITY_CORE_WORKER_ID", raising=False)

    with pytest.raises(HTTPException) as exc:
        require_worker(_request(), x_worker_token="worker-secret")

    assert exc.value.status_code == 503
    assert "ACTIVITY_CORE_WORKER_ID" in exc.value.detail


def test_invalid_worker_token_is_rejected(monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setenv("ACTIVITY_CORE_WORKER_TOKEN", "worker-secret")
    monkeypatch.setenv("ACTIVITY_CORE_WORKER_ID", "rein-aharness@railiance01")

    with pytest.raises(HTTPException) as exc:
        require_worker(_request(), x_worker_token="wrong")

    assert exc.value.status_code == 401


def test_operator_or_sso_is_not_worker_identity(monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.delenv("ACTIVITY_CORE_WORKER_TOKEN", raising=False)
    monkeypatch.delenv("ACTIVITY_CORE_WORKER_ID", raising=False)
    monkeypatch.setenv("ACTIVITY_CORE_OPERATOR_TOKEN", "operator-secret")

    for request, kwargs in (
        (_request({"Remote-User": "alice"}), {}),
        (_request(), {"authorization": "Bearer operator-secret"}),
    ):
        with pytest.raises(HTTPException) as exc:
            require_worker(request, **kwargs)
        assert exc.value.status_code == 503


def test_claim_endpoint_rejects_operator_and_spoofed_identity(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    monkeypatch.setenv("ACTIVITY_CORE_WORKER_TOKEN", "worker-secret")
    monkeypatch.setenv("ACTIVITY_CORE_WORKER_ID", "rein-aharness@railiance01")
    monkeypatch.setenv("ACTIVITY_CORE_OPERATOR_TOKEN", "operator-secret")
    client = _client()
    body = {"worker_id": "rein-aharness@railiance01", "limit": 1}

    operator = client.post(
        "/ops-runs/claim",
        json=body,
        headers={"X-Operator-Token": "operator-secret"},
    )
    spoofed = client.post(
        "/ops-runs/claim",
        json={"worker_id": "another-worker", "limit": 1},
        headers={"X-Worker-Token": "worker-secret"},
    )

    assert operator.status_code == 401
    assert spoofed.status_code == 403


def test_unauthenticated_dev_worker_requires_explicit_opt_in(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    monkeypatch.delenv("ACTIVITY_CORE_WORKER_TOKEN", raising=False)
    monkeypatch.delenv("ACTIVITY_CORE_WORKER_ID", raising=False)
    monkeypatch.delenv("ACTIVITY_CORE_OPERATOR_TOKEN", raising=False)
    monkeypatch.delenv("ACTIVITY_CORE_OPS_ALLOW_UNAUTH_MUTATIONS", raising=False)

    with pytest.raises(HTTPException) as exc:
        require_worker(_request())
    assert exc.value.status_code == 503

    monkeypatch.setenv("ACTIVITY_CORE_OPS_ALLOW_UNAUTH_MUTATIONS", "true")
    assert require_worker(_request()) == "dev:unauth"
    assert bind_worker_id("local-worker", "dev:unauth") == "local-worker"


def test_read_auth_no_longer_defaults_open(monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.delenv("ACTIVITY_CORE_WORKER_TOKEN", raising=False)
    monkeypatch.delenv("ACTIVITY_CORE_WORKER_ID", raising=False)
    monkeypatch.delenv("ACTIVITY_CORE_OPERATOR_TOKEN", raising=False)
    monkeypatch.delenv("ACTIVITY_CORE_OPS_ALLOW_UNAUTH_MUTATIONS", raising=False)

    with pytest.raises(HTTPException) as exc:
        require_worker_or_operator(_request())

    assert exc.value.status_code == 503


def test_close_response_distinguishes_reconciled_and_refusal_codes(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    row = MagicMock()
    monkeypatch.setattr(
        "activity_core.ops_runs_api.ops_run_to_dict",
        lambda value: {"id": "run-1", "state": value.state},
    )
    row.state = "succeeded"

    response = _close_response(CloseOpsRunOutcome("reconciled", row))
    assert response == {
        "id": "run-1",
        "state": "succeeded",
        "close_disposition": "reconciled",
    }

    with pytest.raises(HTTPException) as exc:
        _close_response(CloseOpsRunOutcome("expired_lease", row))
    assert exc.value.status_code == 409
    assert exc.value.detail["code"] == "expired_lease"


def _clear_worker_env(monkeypatch: pytest.MonkeyPatch) -> None:
    for name in (
        "ACTIVITY_CORE_WORKERS",
        "ACTIVITY_CORE_WORKER_ID",
        "ACTIVITY_CORE_WORKER_TOKEN",
        "ACTIVITY_CORE_WORKER_TOKEN_METERED",
        "ACTIVITY_CORE_OPERATOR_TOKEN",
        "ACTIVITY_CORE_OPS_ALLOW_UNAUTH_MUTATIONS",
    ):
        monkeypatch.delenv(name, raising=False)


def _two_workers(monkeypatch: pytest.MonkeyPatch) -> None:
    _clear_worker_env(monkeypatch)
    monkeypatch.setenv(
        "ACTIVITY_CORE_WORKERS",
        "rein-aharness@railiance01=ACTIVITY_CORE_WORKER_TOKEN,"
        "rein-aharness-metered@railiance01=ACTIVITY_CORE_WORKER_TOKEN_METERED",
    )
    monkeypatch.setenv("ACTIVITY_CORE_WORKER_TOKEN", "loop-secret")
    monkeypatch.setenv("ACTIVITY_CORE_WORKER_TOKEN_METERED", "metered-secret")


def test_each_worker_token_binds_exactly_its_own_identity(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    _two_workers(monkeypatch)

    loop = require_worker(_request(), x_worker_token="loop-secret")
    metered = require_worker(_request(), authorization="Bearer metered-secret")

    assert loop == "rein-aharness@railiance01"
    assert metered == "rein-aharness-metered@railiance01"
    with pytest.raises(HTTPException) as exc:
        bind_worker_id("rein-aharness@railiance01", metered)
    assert exc.value.status_code == 403
    with pytest.raises(HTTPException) as exc:
        require_worker(_request(), x_worker_token="unknown")
    assert exc.value.status_code == 401


def test_claim_endpoint_rejects_cross_identity_body(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    _two_workers(monkeypatch)

    response = _client().post(
        "/ops-runs/claim",
        json={"worker_id": "rein-aharness@railiance01", "limit": 1},
        headers={"X-Worker-Token": "metered-secret"},
    )

    assert response.status_code == 403


@pytest.mark.parametrize(
    ("workers", "env", "fragment"),
    [
        ("a=ACTIVITY_CORE_WORKER_TOKEN,b=ACTIVITY_CORE_WORKER_TOKEN_METERED",
         {"ACTIVITY_CORE_WORKER_TOKEN": "same", "ACTIVITY_CORE_WORKER_TOKEN_METERED": "same"},
         "shared"),
        ("a=ACTIVITY_CORE_WORKER_TOKEN,a=ACTIVITY_CORE_WORKER_TOKEN_METERED",
         {"ACTIVITY_CORE_WORKER_TOKEN": "x", "ACTIVITY_CORE_WORKER_TOKEN_METERED": "y"},
         "duplicate"),
        ("a=ACTIVITY_CORE_WORKER_TOKEN,b=ACTIVITY_CORE_WORKER_TOKEN_METERED",
         {"ACTIVITY_CORE_WORKER_TOKEN": "x"},
         "ACTIVITY_CORE_WORKER_TOKEN_METERED"),
        ("a=ACTIVITY_CORE_OPERATOR_TOKEN",
         {"ACTIVITY_CORE_OPERATOR_TOKEN": "op"},
         "invalid"),
        ("a=ACTIVITY_CORE_WORKER_TOKEN",
         {"ACTIVITY_CORE_WORKER_TOKEN": "op", "ACTIVITY_CORE_OPERATOR_TOKEN": "op"},
         "operator"),
        ("a=ACTIVITY_CORE_WORKER_TOKEN",
         {"ACTIVITY_CORE_WORKER_TOKEN": "x", "ACTIVITY_CORE_WORKER_ID": "b"},
         "not listed"),
    ],
)
def test_ambiguous_worker_config_fails_closed(
    monkeypatch: pytest.MonkeyPatch,
    workers: str,
    env: dict[str, str],
    fragment: str,
) -> None:
    _clear_worker_env(monkeypatch)
    monkeypatch.setenv("ACTIVITY_CORE_WORKERS", workers)
    for name, value in env.items():
        monkeypatch.setenv(name, value)

    with pytest.raises(HTTPException) as exc:
        require_worker(_request(), x_worker_token="x")

    assert exc.value.status_code == 503
    assert fragment in exc.value.detail


def test_broken_worker_config_does_not_lock_out_operator(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    _clear_worker_env(monkeypatch)
    monkeypatch.setenv("ACTIVITY_CORE_WORKERS", "a=ACTIVITY_CORE_WORKER_TOKEN_MISSING")
    monkeypatch.setenv("ACTIVITY_CORE_OPERATOR_TOKEN", "operator-secret")

    principal = require_worker_or_operator(
        _request(), authorization="Bearer operator-secret"
    )

    assert principal == "operator:token"