tegwick
672cf4da6e
All checks were successful
Build and Publish Container Image / build-and-push (push) Successful in 37s
Engine/PIP declaration is now checkable (layer.yaml plus a Tooling-client scan). Writes persist a decision record or the published fail-closed stance, live-lookup freshness is published, events_for is tenant-scoped, and mutation evidence drains to audit-core from a local outbox without blocking the mutation. Sender registration is requested as AUDIT-IN-0002. Boundary-contract amendment is requested as NET-IN-0002. Assistant: grok Assistant-Session: 01a04cea-e5e8-7081-a0fc-808ebbc35fa9
122 lines
3.9 KiB
Python
122 lines
3.9 KiB
Python
from pathlib import Path
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from tenant_engine.flex_auth import CheckRequest, FlexAuthCheckClient, new_request_id
|
|
|
|
|
|
def _request() -> CheckRequest:
|
|
return CheckRequest(
|
|
request_id=new_request_id(),
|
|
tenant="tenant:friendly:binky",
|
|
subject_id="tenant-engine",
|
|
subject_type="service",
|
|
action="tenant.create",
|
|
resource_id="t-1",
|
|
resource_type="tenant",
|
|
)
|
|
|
|
|
|
def _client(handler) -> FlexAuthCheckClient:
|
|
return FlexAuthCheckClient(
|
|
base_url="https://flex-auth.example.test",
|
|
timeout_seconds=1,
|
|
transport=httpx.MockTransport(handler),
|
|
)
|
|
|
|
|
|
def test_allow_effect_authorizes() -> None:
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
return httpx.Response(
|
|
200,
|
|
json={"id": "d-1", "effect": "allow", "resource": {}, "subject": {}, "provenance": {}},
|
|
)
|
|
|
|
assert _client(handler).is_allowed(_request()) is True
|
|
|
|
|
|
@pytest.mark.parametrize("effect", ["deny", "redact", "audit_only", "not_applicable"])
|
|
def test_non_allow_effects_deny(effect: str) -> None:
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
return httpx.Response(
|
|
200,
|
|
json={"id": "d-1", "effect": effect, "resource": {}, "subject": {}, "provenance": {}},
|
|
)
|
|
|
|
assert _client(handler).is_allowed(_request()) is False
|
|
|
|
|
|
def test_non_200_status_denies() -> None:
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
return httpx.Response(500, json={"error": "internal"})
|
|
|
|
assert _client(handler).is_allowed(_request()) is False
|
|
|
|
|
|
def test_malformed_json_body_denies() -> None:
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
return httpx.Response(200, content=b"not json")
|
|
|
|
assert _client(handler).is_allowed(_request()) is False
|
|
|
|
|
|
def test_non_object_json_body_denies() -> None:
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
return httpx.Response(200, json=["not", "an", "object"])
|
|
|
|
assert _client(handler).is_allowed(_request()) is False
|
|
|
|
|
|
def test_connection_failure_denies() -> None:
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
raise httpx.ConnectError("connection refused", request=request)
|
|
|
|
assert _client(handler).is_allowed(_request()) is False
|
|
|
|
|
|
def test_timeout_denies() -> None:
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
raise httpx.TimeoutException("timed out", request=request)
|
|
|
|
assert _client(handler).is_allowed(_request()) is False
|
|
|
|
|
|
def test_request_body_matches_schema_shape() -> None:
|
|
seen: dict[str, object] = {}
|
|
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
import json
|
|
|
|
seen.update(json.loads(request.content))
|
|
return httpx.Response(
|
|
200,
|
|
json={"id": "d-1", "effect": "allow", "resource": {}, "subject": {}, "provenance": {}},
|
|
)
|
|
|
|
_client(handler).is_allowed(_request())
|
|
|
|
assert seen["tenant"] == "tenant:friendly:binky"
|
|
assert seen["action"] == "tenant.create"
|
|
assert seen["subject"] == {"id": "tenant-engine", "type": "service"}
|
|
assert seen["resource"] == {"id": "t-1", "type": "tenant", "system": "tenant-engine"}
|
|
|
|
|
|
def test_rotating_caller_token_is_read_for_each_check(tmp_path: Path) -> None:
|
|
token_file = tmp_path / "token"
|
|
token_file.write_text("token-one\n")
|
|
seen: list[str] = []
|
|
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
seen.append(request.headers["authorization"])
|
|
return httpx.Response(200, json={"id": "d-1", "effect": "allow"})
|
|
|
|
client = FlexAuthCheckClient(
|
|
base_url="https://flex-auth.example.test",
|
|
transport=httpx.MockTransport(handler),
|
|
bearer_token_file=str(token_file),
|
|
)
|
|
assert client.is_allowed(_request()) is True
|
|
token_file.write_text("token-two\n")
|
|
assert client.is_allowed(_request()) is True
|
|
assert seen == ["Bearer token-one", "Bearer token-two"]
|