tegwick
f885697e96
Mark workplan active. Add Traefik ForwardAuth middleware and Ingress manifests for activity.coulomb.social and activity-temporal.coulomb.social. Prefer Authelia SSO identity for ops mutations; document DNS gate and fleet pattern (docs/ops-sso-access.md).
96 lines
3.3 KiB
Python
96 lines
3.3 KiB
Python
"""Unit tests for operator token + SSO auth (ACTIVITY-WP-0024/0025)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from unittest.mock import MagicMock
|
|
|
|
import pytest
|
|
from fastapi import HTTPException
|
|
|
|
from activity_core.ops_auth import (
|
|
extract_operator_token,
|
|
extract_sso_principal,
|
|
operator_token_configured,
|
|
require_operator,
|
|
verify_operator_token,
|
|
)
|
|
|
|
|
|
def test_extract_operator_token_prefers_header() -> None:
|
|
assert (
|
|
extract_operator_token(
|
|
x_operator_token="abc",
|
|
authorization="Bearer other",
|
|
)
|
|
== "abc"
|
|
)
|
|
|
|
|
|
def test_extract_operator_token_bearer() -> None:
|
|
assert extract_operator_token(authorization="Bearer secret-token") == "secret-token"
|
|
|
|
|
|
def test_verify_requires_config_when_fail_closed(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.delenv("ACTIVITY_CORE_OPERATOR_TOKEN", raising=False)
|
|
monkeypatch.delenv("ACTIVITY_CORE_OPS_ALLOW_UNAUTH_MUTATIONS", raising=False)
|
|
with pytest.raises(HTTPException) as exc:
|
|
verify_operator_token(None)
|
|
assert exc.value.status_code == 403
|
|
|
|
|
|
def test_verify_allows_anonymous_dev(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.delenv("ACTIVITY_CORE_OPERATOR_TOKEN", raising=False)
|
|
monkeypatch.setenv("ACTIVITY_CORE_OPS_ALLOW_UNAUTH_MUTATIONS", "1")
|
|
assert verify_operator_token(None) == "anonymous-dev"
|
|
|
|
|
|
def test_verify_token_match(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setenv("ACTIVITY_CORE_OPERATOR_TOKEN", "correct-horse")
|
|
assert verify_operator_token("correct-horse") == "operator-token"
|
|
with pytest.raises(HTTPException) as exc:
|
|
verify_operator_token("wrong")
|
|
assert exc.value.status_code == 401
|
|
with pytest.raises(HTTPException) as exc2:
|
|
verify_operator_token(None)
|
|
assert exc2.value.status_code == 401
|
|
|
|
|
|
def test_operator_token_configured(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.delenv("ACTIVITY_CORE_OPERATOR_TOKEN", raising=False)
|
|
assert operator_token_configured() is False
|
|
monkeypatch.setenv("ACTIVITY_CORE_OPERATOR_TOKEN", "x")
|
|
assert operator_token_configured() is True
|
|
|
|
|
|
def test_extract_sso_principal_remote_user() -> None:
|
|
req = MagicMock()
|
|
req.headers = {"Remote-User": "bernd", "Remote-Email": "bernd@example.com"}
|
|
# MagicMock headers.get needs side_effect
|
|
headers = {"Remote-User": "bernd", "Remote-Email": "bernd@example.com"}
|
|
|
|
class H(dict):
|
|
def get(self, key, default=None): # type: ignore[no-untyped-def]
|
|
for k, v in self.items():
|
|
if k.lower() == str(key).lower():
|
|
return v
|
|
return default
|
|
|
|
req.headers = H(headers)
|
|
assert extract_sso_principal(req) == "bernd"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_require_operator_prefers_sso(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setenv("ACTIVITY_CORE_OPERATOR_TOKEN", "tok")
|
|
|
|
class H(dict):
|
|
def get(self, key, default=None): # type: ignore[no-untyped-def]
|
|
for k, v in self.items():
|
|
if k.lower() == str(key).lower():
|
|
return v
|
|
return default
|
|
|
|
req = MagicMock()
|
|
req.headers = H({"Remote-User": "alice"})
|
|
principal = await require_operator(req, x_operator_token=None, authorization=None)
|
|
assert principal == "sso:alice"
|