1192 lines
45 KiB
Python
1192 lines
45 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import uuid
|
|
from collections.abc import Callable, Iterator
|
|
from dataclasses import dataclass
|
|
from datetime import UTC, datetime
|
|
from unittest.mock import Mock
|
|
|
|
import pytest
|
|
from fastapi import FastAPI, HTTPException, Request
|
|
from fastapi.routing import APIRoute
|
|
from fastapi.testclient import TestClient
|
|
from pydantic import SecretStr
|
|
from sqlalchemy import create_engine, text
|
|
from sqlalchemy.exc import IntegrityError, OperationalError
|
|
from sqlalchemy.orm import Session
|
|
from sqlalchemy.pool import StaticPool
|
|
|
|
from modelforge_api.api.authorization import (
|
|
AccessBoundary,
|
|
Principal,
|
|
PrincipalRole,
|
|
access_boundary_for_request,
|
|
authenticate_operator_principal,
|
|
authenticate_operator_token,
|
|
require_admin,
|
|
require_operator,
|
|
require_viewer,
|
|
)
|
|
from modelforge_api.api.request_limits import MAX_CONSECUTIVE_EMPTY_REQUEST_EVENTS
|
|
from modelforge_api.api.routes.acquisition import get_acquisition_service
|
|
from modelforge_api.api.routes.agent import authenticated_node, get_agent_service
|
|
from modelforge_api.api.routes.hardware import get_hardware_service
|
|
from modelforge_api.api.routes.registry import get_registry_service
|
|
from modelforge_api.api.routes.runtime import get_runtime_service
|
|
from modelforge_api.api.routes.serving import (
|
|
authenticate_document_ocr_client,
|
|
authenticate_rag_embedding_client,
|
|
authenticate_rag_reranking_client,
|
|
authenticate_speech_transcription_client,
|
|
authenticate_vision_embedding_client,
|
|
get_serving_service,
|
|
)
|
|
from modelforge_api.domain.serving import ServiceClientCreate
|
|
from modelforge_api.main import (
|
|
app,
|
|
correlation_middleware,
|
|
interactive_api_enabled_for,
|
|
)
|
|
from modelforge_api.main import (
|
|
settings as application_settings,
|
|
)
|
|
from modelforge_api.persistence.models import Base, ComputeNode, NodeCredential
|
|
from modelforge_api.services.manifest_registry import ManifestRegistry
|
|
from modelforge_api.services.node_agent import NodeAgentService, secret_hash
|
|
from modelforge_api.services.registry import RegistryService
|
|
from modelforge_api.services.serving import ServingService
|
|
from modelforge_api.services.transient_payloads import MemoryPayloadStore
|
|
from modelforge_api.settings import Settings, get_settings
|
|
|
|
|
|
def _test_admin_token() -> str:
|
|
return "rc00-legacy-admin"
|
|
|
|
|
|
ADMIN_TOKEN = _test_admin_token()
|
|
ADMIN = {"X-ModelForge-Admin-Token": ADMIN_TOKEN}
|
|
ALLOWED_ORIGIN = application_settings.cors_origin_list[0]
|
|
DISALLOWED_ORIGIN = "https://not-allowed.invalid"
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class MachineBoundaryHarness:
|
|
settings: Settings
|
|
node_service: NodeAgentService
|
|
capability_service: ServingService
|
|
node_credential: str
|
|
revoked_node_credential: str
|
|
wrong_scope_node_credential: str
|
|
capability_credential: str
|
|
revoked_capability_credential: str
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class AsgiProbeResult:
|
|
status_code: int
|
|
headers: dict[str, str]
|
|
body: bytes
|
|
body_receive_calls: int
|
|
body_bytes_produced: int
|
|
|
|
|
|
async def _asgi_request(
|
|
method: str,
|
|
path: str,
|
|
*,
|
|
authorization: str | None = None,
|
|
operator_token: str | None = None,
|
|
body: bytes | None = None,
|
|
streamed_body_bytes: int | None = None,
|
|
stream_chunk_bytes: int = 32_768,
|
|
content_length: int | None = None,
|
|
fail_on_receive: bool = False,
|
|
request_messages: list[dict[str, object]] | None = None,
|
|
) -> AsgiProbeResult:
|
|
request_headers = [
|
|
(b"host", b"testserver"),
|
|
(b"content-type", b"application/json"),
|
|
(b"origin", ALLOWED_ORIGIN.encode()),
|
|
]
|
|
if authorization is not None:
|
|
request_headers.append((b"authorization", authorization.encode()))
|
|
if operator_token is not None:
|
|
request_headers.append((b"x-modelforge-admin-token", operator_token.encode()))
|
|
if content_length is not None:
|
|
request_headers.append((b"content-length", str(content_length).encode()))
|
|
|
|
body_receive_calls = 0
|
|
body_bytes_produced = 0
|
|
fixed_body_sent = False
|
|
request_message_index = 0
|
|
|
|
async def receive() -> dict[str, object]:
|
|
nonlocal body_receive_calls, body_bytes_produced, fixed_body_sent, request_message_index
|
|
if fail_on_receive:
|
|
raise AssertionError("request body receive must not be called")
|
|
if request_messages is not None:
|
|
if request_message_index >= len(request_messages):
|
|
raise AssertionError("request body receive continued beyond the supplied frame budget")
|
|
message = request_messages[request_message_index]
|
|
request_message_index += 1
|
|
if message["type"] == "http.request":
|
|
body_receive_calls += 1
|
|
body_bytes_produced += len(message.get("body", b""))
|
|
return message
|
|
if streamed_body_bytes is not None:
|
|
if body_bytes_produced >= streamed_body_bytes:
|
|
return {"type": "http.disconnect"}
|
|
amount = min(stream_chunk_bytes, streamed_body_bytes - body_bytes_produced)
|
|
body_receive_calls += 1
|
|
body_bytes_produced += amount
|
|
return {
|
|
"type": "http.request",
|
|
"body": b"x" * amount,
|
|
"more_body": body_bytes_produced < streamed_body_bytes,
|
|
}
|
|
if not fixed_body_sent:
|
|
fixed_body_sent = True
|
|
payload = body or b""
|
|
body_receive_calls += 1
|
|
body_bytes_produced += len(payload)
|
|
return {"type": "http.request", "body": payload, "more_body": False}
|
|
return {"type": "http.disconnect"}
|
|
|
|
sent: list[dict[str, object]] = []
|
|
|
|
async def send(message: dict[str, object]) -> None:
|
|
sent.append(message)
|
|
|
|
scope: dict[str, object] = {
|
|
"type": "http",
|
|
"asgi": {"version": "3.0"},
|
|
"http_version": "1.1",
|
|
"method": method,
|
|
"scheme": "http",
|
|
"path": path,
|
|
"raw_path": path.encode(),
|
|
"query_string": b"",
|
|
"root_path": "",
|
|
"headers": request_headers,
|
|
"client": ("testclient", 50000),
|
|
"server": ("testserver", 80),
|
|
}
|
|
await app(scope, receive, send)
|
|
start = next(message for message in sent if message["type"] == "http.response.start")
|
|
response_headers = {
|
|
name.decode().lower(): value.decode() for name, value in start.get("headers", [])
|
|
}
|
|
response_body = b"".join(
|
|
message.get("body", b"") for message in sent if message["type"] == "http.response.body"
|
|
)
|
|
return AsgiProbeResult(
|
|
status_code=int(start["status"]),
|
|
headers=response_headers,
|
|
body=response_body,
|
|
body_receive_calls=body_receive_calls,
|
|
body_bytes_produced=body_bytes_produced,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def machine_boundary_harness() -> Iterator[MachineBoundaryHarness]:
|
|
engine = create_engine(
|
|
"sqlite+pysqlite:///:memory:",
|
|
connect_args={"check_same_thread": False},
|
|
poolclass=StaticPool,
|
|
)
|
|
Base.metadata.create_all(engine)
|
|
settings = Settings(
|
|
_env_file=None,
|
|
operator_api_key=SecretStr(ADMIN_TOKEN),
|
|
gateway_max_payload_bytes=65_536,
|
|
node_agent_max_payload_bytes=131_072,
|
|
control_plane_max_payload_bytes=65_536,
|
|
)
|
|
with Session(engine) as session:
|
|
valid_node = ComputeNode(key="prebody-valid", hostname="prebody-valid")
|
|
revoked_node = ComputeNode(key="prebody-revoked", hostname="prebody-revoked")
|
|
wrong_scope_node = ComputeNode(
|
|
key="prebody-wrong-scope", hostname="prebody-wrong-scope"
|
|
)
|
|
session.add_all([valid_node, revoked_node, wrong_scope_node])
|
|
session.flush()
|
|
valid_node_credential_id = uuid.uuid4()
|
|
valid_node_credential = f"mfnode_{valid_node_credential_id}_valid-prebody-secret"
|
|
revoked_node_credential_id = uuid.uuid4()
|
|
revoked_node_credential = f"mfnode_{revoked_node_credential_id}_revoked-prebody-secret"
|
|
wrong_scope_node_credential_id = uuid.uuid4()
|
|
wrong_scope_node_credential = (
|
|
f"mfnode_{wrong_scope_node_credential_id}_wrong-scope-prebody-secret"
|
|
)
|
|
session.add_all(
|
|
[
|
|
NodeCredential(
|
|
id=valid_node_credential_id,
|
|
compute_node_id=valid_node.id,
|
|
secret_hash=secret_hash(valid_node_credential),
|
|
),
|
|
NodeCredential(
|
|
id=revoked_node_credential_id,
|
|
compute_node_id=revoked_node.id,
|
|
secret_hash=secret_hash(revoked_node_credential),
|
|
revoked_at=datetime.now(UTC),
|
|
),
|
|
NodeCredential(
|
|
id=wrong_scope_node_credential_id,
|
|
compute_node_id=wrong_scope_node.id,
|
|
secret_hash=secret_hash(wrong_scope_node_credential),
|
|
),
|
|
]
|
|
)
|
|
session.commit()
|
|
session.execute(text("PRAGMA ignore_check_constraints = ON"))
|
|
wrong_scope_credential = session.get(
|
|
NodeCredential,
|
|
wrong_scope_node_credential_id,
|
|
)
|
|
assert wrong_scope_credential is not None
|
|
wrong_scope_credential.scope = "node.enroll"
|
|
session.commit()
|
|
session.execute(text("PRAGMA ignore_check_constraints = OFF"))
|
|
|
|
node_service = NodeAgentService(session, settings)
|
|
capability_service = ServingService(
|
|
session,
|
|
settings,
|
|
ManifestRegistry(),
|
|
MemoryPayloadStore(),
|
|
)
|
|
capability_service.ensure_contract()
|
|
valid_client = capability_service.create_client(
|
|
ServiceClientCreate(
|
|
name="prebody-valid-client",
|
|
allowed_capabilities=["rag.embedding@1"],
|
|
)
|
|
)
|
|
revoked_client = capability_service.create_client(
|
|
ServiceClientCreate(
|
|
name="prebody-revoked-client",
|
|
allowed_capabilities=["rag.embedding@1"],
|
|
)
|
|
)
|
|
capability_service.revoke_client_credential(revoked_client.id)
|
|
|
|
app.dependency_overrides[get_settings] = lambda: settings
|
|
app.dependency_overrides[get_agent_service] = lambda: node_service
|
|
app.dependency_overrides[get_serving_service] = lambda: capability_service
|
|
try:
|
|
yield MachineBoundaryHarness(
|
|
settings=settings,
|
|
node_service=node_service,
|
|
capability_service=capability_service,
|
|
node_credential=valid_node_credential,
|
|
revoked_node_credential=revoked_node_credential,
|
|
wrong_scope_node_credential=wrong_scope_node_credential,
|
|
capability_credential=valid_client.credential,
|
|
revoked_capability_credential=revoked_client.credential,
|
|
)
|
|
finally:
|
|
app.dependency_overrides.clear()
|
|
|
|
|
|
@pytest.fixture
|
|
def operator_client() -> Iterator[TestClient]:
|
|
settings = Settings(_env_file=None, operator_api_key=SecretStr(ADMIN_TOKEN))
|
|
app.dependency_overrides[get_settings] = lambda: settings
|
|
# Invalid-body and authentication-boundary requests must not need infrastructure. A successful
|
|
# control overrides the registry service with a real in-memory repository in its own test.
|
|
app.dependency_overrides[get_acquisition_service] = object
|
|
app.dependency_overrides[get_hardware_service] = object
|
|
app.dependency_overrides[get_registry_service] = object
|
|
app.dependency_overrides[get_runtime_service] = object
|
|
try:
|
|
yield TestClient(app)
|
|
finally:
|
|
app.dependency_overrides.clear()
|
|
|
|
|
|
TARGET_MUTATION_POLICIES = {
|
|
# Acquisition: operators prepare and run work; approval is an administrative decision.
|
|
("POST", "/api/v1/discovery/search"): "operator",
|
|
("POST", "/api/v1/models/{model_id}/refresh-upstream"): "operator",
|
|
("POST", "/api/v1/download-plans"): "operator",
|
|
("POST", "/api/v1/download-plans/{plan_id}/approve"): "admin",
|
|
("POST", "/api/v1/download-plans/{plan_id}/execute"): "operator",
|
|
("POST", "/api/v1/artifact-jobs/{job_id}/cancel"): "operator",
|
|
("POST", "/api/v1/artifact-jobs/{job_id}/retry"): "operator",
|
|
# Hardware refresh is observable operator work; persisted node administration remains admin.
|
|
("POST", "/api/v1/hardware/refresh"): "operator",
|
|
# Registry creation/inspection is operator work. Destructive lifecycle and roots are admin.
|
|
("POST", "/api/v1/models"): "operator",
|
|
("PATCH", "/api/v1/models/{model_id}"): "operator",
|
|
("POST", "/api/v1/models/{model_id}/deprecate"): "admin",
|
|
("POST", "/api/v1/models/{model_id}/archive"): "admin",
|
|
("DELETE", "/api/v1/models/{model_id}"): "admin",
|
|
("POST", "/api/v1/models/{model_id}/revisions"): "operator",
|
|
("DELETE", "/api/v1/revisions/{revision_id}"): "admin",
|
|
("POST", "/api/v1/revisions/{revision_id}/artifacts"): "operator",
|
|
("DELETE", "/api/v1/artifacts/{artifact_id}"): "admin",
|
|
("POST", "/api/v1/artifacts/{artifact_id}/verify"): "operator",
|
|
("POST", "/api/v1/derived-artifacts"): "operator",
|
|
("DELETE", "/api/v1/derived-artifacts/{artifact_id}"): "admin",
|
|
("POST", "/api/v1/storage-roots"): "admin",
|
|
("PATCH", "/api/v1/storage-roots/{root_id}"): "admin",
|
|
("POST", "/api/v1/storage-roots/{root_id}/observations"): "operator",
|
|
("GET", "/api/v1/storage-roots/{root_id}/capacity"): "operator",
|
|
("DELETE", "/api/v1/storage-roots/{root_id}"): "admin",
|
|
# Lab construction/probes are operator work; granting execution approval is admin.
|
|
("POST", "/api/v1/runtime-environments"): "operator",
|
|
("POST", "/api/v1/runtime-profiles"): "operator",
|
|
(
|
|
"POST",
|
|
"/api/v1/artifact-sets/{artifact_set_id}/compatibility-assessments",
|
|
): "operator",
|
|
("POST", "/api/v1/artifact-sets/{artifact_set_id}/execution-approvals"): "admin",
|
|
("POST", "/api/v1/runtime-probes"): "operator",
|
|
("POST", "/api/v1/runtime-probes/{probe_id}/cancel"): "operator",
|
|
}
|
|
|
|
|
|
DEPENDENCY_POLICIES = {
|
|
require_viewer: "viewer",
|
|
require_operator: "operator",
|
|
require_admin: "admin",
|
|
authenticated_node: "node",
|
|
authenticate_rag_embedding_client: "capability_client",
|
|
authenticate_rag_reranking_client: "capability_client",
|
|
authenticate_document_ocr_client: "capability_client",
|
|
authenticate_vision_embedding_client: "capability_client",
|
|
authenticate_speech_transcription_client: "capability_client",
|
|
}
|
|
|
|
CAPABILITY_AUTHENTICATORS = {
|
|
authenticate_rag_embedding_client,
|
|
authenticate_rag_reranking_client,
|
|
authenticate_document_ocr_client,
|
|
authenticate_vision_embedding_client,
|
|
authenticate_speech_transcription_client,
|
|
}
|
|
|
|
CAPABILITY_AUTHENTICATOR_SCOPES = {
|
|
authenticate_rag_embedding_client: "rag.embedding@1",
|
|
authenticate_rag_reranking_client: "rag.reranking@1",
|
|
authenticate_document_ocr_client: "document.ocr@1",
|
|
authenticate_vision_embedding_client: "vision.embedding@1",
|
|
authenticate_speech_transcription_client: "speech.transcription@1",
|
|
}
|
|
|
|
EXPECTED_CAPABILITY_AUTHENTICATORS = {
|
|
("POST", "/api/v1/capabilities/rag.embedding@1/invoke"): authenticate_rag_embedding_client,
|
|
("POST", "/api/v1/capabilities/rag.reranking@1/invoke"): authenticate_rag_reranking_client,
|
|
("POST", "/api/v1/capabilities/document.ocr@1/invoke"): authenticate_document_ocr_client,
|
|
(
|
|
"POST",
|
|
"/api/v1/capabilities/vision.embedding@1/invoke",
|
|
): authenticate_vision_embedding_client,
|
|
(
|
|
"POST",
|
|
"/api/v1/capabilities/speech.transcription@1/invoke",
|
|
): authenticate_speech_transcription_client,
|
|
(
|
|
"POST",
|
|
"/api/v1/capability-experiments/{route_key}/invoke",
|
|
): authenticate_rag_embedding_client,
|
|
("POST", "/v1/embeddings"): authenticate_rag_embedding_client,
|
|
}
|
|
|
|
|
|
def route_policy(route: APIRoute, method: str) -> str | None:
|
|
policies = {
|
|
DEPENDENCY_POLICIES[dependency.call]
|
|
for dependency in route.dependant.dependencies
|
|
if dependency.call in DEPENDENCY_POLICIES
|
|
}
|
|
human_policies = policies & {"viewer", "operator", "admin"}
|
|
machine_policies = policies - human_policies
|
|
if machine_policies and human_policies:
|
|
raise AssertionError(f"conflicting policies for {method} {route.path}: {policies}")
|
|
if machine_policies:
|
|
if len(machine_policies) > 1:
|
|
raise AssertionError(f"conflicting policies for {method} {route.path}: {policies}")
|
|
return machine_policies.pop()
|
|
if human_policies:
|
|
return max(human_policies, key={"viewer": 0, "operator": 1, "admin": 2}.__getitem__)
|
|
boundary = access_boundary_for_request(method, route.path)
|
|
if boundary is AccessBoundary.PUBLIC:
|
|
return "public"
|
|
if boundary is AccessBoundary.SINGLE_USE_ENROLLMENT:
|
|
return "single_use_enrollment"
|
|
return None
|
|
|
|
|
|
def test_every_route_has_one_explicit_access_policy() -> None:
|
|
unclassified: list[tuple[str, str]] = []
|
|
for route in app.routes:
|
|
if not isinstance(route, APIRoute):
|
|
continue
|
|
for method in route.methods:
|
|
if route_policy(route, method) is None:
|
|
unclassified.append((method, route.path))
|
|
assert unclassified == []
|
|
|
|
|
|
def test_every_control_plane_route_has_an_explicit_human_role_dependency() -> None:
|
|
missing: list[tuple[str, str]] = []
|
|
for route in app.routes:
|
|
if not isinstance(route, APIRoute):
|
|
continue
|
|
dependencies = {dependency.call for dependency in route.dependant.dependencies}
|
|
for method in route.methods:
|
|
if access_boundary_for_request(
|
|
method, route.path
|
|
) is AccessBoundary.CONTROL_PLANE and not dependencies.intersection(
|
|
{require_viewer, require_operator, require_admin}
|
|
):
|
|
missing.append((method, route.path))
|
|
assert missing == []
|
|
|
|
|
|
def test_only_contractual_routes_bypass_the_control_plane_boundary() -> None:
|
|
boundaries = {
|
|
(method, route.path): access_boundary_for_request(method, route.path)
|
|
for route in app.routes
|
|
if isinstance(route, APIRoute)
|
|
for method in route.methods
|
|
}
|
|
assert {key for key, boundary in boundaries.items() if boundary is AccessBoundary.PUBLIC} == {
|
|
("GET", "/"),
|
|
("GET", "/api/v1/health/live"),
|
|
("GET", "/api/v1/health/ready"),
|
|
("GET", "/api/v1/version"),
|
|
}
|
|
assert {
|
|
key
|
|
for key, boundary in boundaries.items()
|
|
if boundary is AccessBoundary.SINGLE_USE_ENROLLMENT
|
|
} == {("POST", "/api/v1/agent/enroll")}
|
|
assert {key for key, boundary in boundaries.items() if boundary is AccessBoundary.NODE} == {
|
|
("POST", "/api/v1/agent/heartbeat"),
|
|
("PUT", "/api/v1/agent/inventory"),
|
|
("PUT", "/api/v1/agent/telemetry"),
|
|
("GET", "/api/v1/agent/artifact-jobs/next"),
|
|
("POST", "/api/v1/agent/artifact-jobs/{job_id}/progress"),
|
|
("POST", "/api/v1/agent/artifact-jobs/{job_id}/complete"),
|
|
("POST", "/api/v1/agent/artifact-jobs/{job_id}/fail"),
|
|
("GET", "/api/v1/agent/runtime-probes/next"),
|
|
("POST", "/api/v1/agent/runtime-probes/{probe_id}/progress"),
|
|
("POST", "/api/v1/agent/runtime-probes/{probe_id}/complete"),
|
|
("POST", "/api/v1/agent/runtime-probes/{probe_id}/fail"),
|
|
("GET", "/api/v1/agent/serving-jobs/next"),
|
|
("POST", "/api/v1/agent/serving-jobs/{job_id}/complete"),
|
|
("POST", "/api/v1/agent/serving-jobs/{job_id}/fail"),
|
|
("POST", "/api/v1/agent/serving-state"),
|
|
}
|
|
assert {
|
|
key for key, boundary in boundaries.items() if boundary is AccessBoundary.CAPABILITY_CLIENT
|
|
} == set(EXPECTED_CAPABILITY_AUTHENTICATORS)
|
|
|
|
|
|
def test_machine_routes_have_direct_authentication_dependency_evidence() -> None:
|
|
missing: list[tuple[str, str, str]] = []
|
|
for route in app.routes:
|
|
if not isinstance(route, APIRoute):
|
|
continue
|
|
dependency_calls = {dependency.call for dependency in route.dependant.dependencies}
|
|
for method in route.methods:
|
|
boundary = access_boundary_for_request(method, route.path)
|
|
if boundary is AccessBoundary.NODE and authenticated_node not in dependency_calls:
|
|
missing.append((method, route.path, "authenticated_node"))
|
|
if boundary is AccessBoundary.CAPABILITY_CLIENT and not dependency_calls.intersection(
|
|
CAPABILITY_AUTHENTICATORS
|
|
):
|
|
missing.append((method, route.path, "capability_authenticator"))
|
|
assert missing == []
|
|
|
|
|
|
def test_each_capability_route_authenticates_its_exact_required_scope() -> None:
|
|
routes = {
|
|
(method, route.path): route
|
|
for route in app.routes
|
|
if isinstance(route, APIRoute)
|
|
for method in route.methods
|
|
}
|
|
for key, expected_authenticator in EXPECTED_CAPABILITY_AUTHENTICATORS.items():
|
|
dependency_calls = {dependency.call for dependency in routes[key].dependant.dependencies}
|
|
assert dependency_calls.intersection(CAPABILITY_AUTHENTICATORS) == {expected_authenticator}
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("authenticator", "required_scope"),
|
|
CAPABILITY_AUTHENTICATOR_SCOPES.items(),
|
|
)
|
|
def test_capability_authenticators_enforce_the_declared_scope(
|
|
authenticator: Callable[[Request, ServingService, str | None], object],
|
|
required_scope: str,
|
|
) -> None:
|
|
request = Request({"type": "http", "method": "POST", "path": "/", "headers": []})
|
|
service = Mock(spec=ServingService)
|
|
authenticated_client = object()
|
|
service.authenticate.return_value = authenticated_client
|
|
|
|
result = authenticator(request, service, "Bearer scoped-project-credential")
|
|
|
|
assert result is authenticated_client
|
|
service.authenticate.assert_called_once_with("Bearer scoped-project-credential", required_scope)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("method", "path", "machine_credential"),
|
|
[
|
|
(
|
|
"GET",
|
|
"/api/v1/agent/future-route-probe",
|
|
"Bearer mfnode_future-route-must-not-open-the-boundary",
|
|
),
|
|
(
|
|
"POST",
|
|
"/api/v1/capabilities/future@1/invoke",
|
|
"Bearer mfsvc_future-route-must-not-open-the-boundary",
|
|
),
|
|
],
|
|
)
|
|
def test_future_machine_looking_routes_fail_closed_without_dependency_evidence(
|
|
method: str, path: str, machine_credential: str
|
|
) -> None:
|
|
probe_app = FastAPI(docs_url=None, redoc_url=None, openapi_url=None)
|
|
probe_app.middleware("http")(correlation_middleware)
|
|
probe_app.add_api_route(path, lambda: {"unsafe": True}, methods=[method])
|
|
probe_app.dependency_overrides[get_settings] = lambda: Settings(
|
|
_env_file=None,
|
|
operator_api_key=SecretStr(ADMIN_TOKEN),
|
|
)
|
|
|
|
probe_route = next(
|
|
route for route in probe_app.routes if isinstance(route, APIRoute) and route.path == path
|
|
)
|
|
assert access_boundary_for_request(method, path) is AccessBoundary.CONTROL_PLANE
|
|
assert route_policy(probe_route, method) is None
|
|
|
|
response = TestClient(probe_app).request(
|
|
method,
|
|
path,
|
|
headers={"Authorization": machine_credential},
|
|
)
|
|
assert response.status_code == 401
|
|
assert response.json()["error"]["message"] == "invalid operator credential"
|
|
|
|
|
|
def test_interactive_api_schema_is_enabled_only_for_development() -> None:
|
|
assert interactive_api_enabled_for("development") is True
|
|
assert interactive_api_enabled_for("test") is False
|
|
assert interactive_api_enabled_for("production") is False
|
|
|
|
|
|
def test_audited_mutations_have_the_intended_operator_or_admin_policy() -> None:
|
|
routes = {
|
|
(method, route.path): route_policy(route, method)
|
|
for route in app.routes
|
|
if isinstance(route, APIRoute)
|
|
for method in route.methods
|
|
}
|
|
assert {key: routes[key] for key in TARGET_MUTATION_POLICIES} == TARGET_MUTATION_POLICIES
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("path", "body"),
|
|
[
|
|
("/api/v1/models", {}),
|
|
("/api/v1/download-plans", {}),
|
|
("/api/v1/runtime-environments", {}),
|
|
],
|
|
)
|
|
def test_authentication_precedes_body_validation(
|
|
operator_client: TestClient, path: str, body: dict[str, object]
|
|
) -> None:
|
|
unauthenticated = operator_client.post(path, json=body)
|
|
assert unauthenticated.status_code == 401
|
|
assert unauthenticated.json()["error"]["message"] == "invalid operator credential"
|
|
|
|
authenticated = operator_client.post(path, headers=ADMIN, json=body)
|
|
assert authenticated.status_code == 422
|
|
assert authenticated.json()["error"]["code"] == "request_validation_failed"
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"operator_headers",
|
|
[
|
|
{},
|
|
{"X-ModelForge-Admin-Token": "wrong-operator-secret"},
|
|
],
|
|
)
|
|
def test_allowed_browser_origin_receives_cors_headers_on_pre_body_401(
|
|
operator_client: TestClient,
|
|
operator_headers: dict[str, str],
|
|
) -> None:
|
|
response = operator_client.post(
|
|
"/api/v1/models",
|
|
headers={"Origin": ALLOWED_ORIGIN, **operator_headers},
|
|
json={},
|
|
)
|
|
assert response.status_code == 401
|
|
assert response.headers["access-control-allow-origin"] == ALLOWED_ORIGIN
|
|
assert response.headers["access-control-allow-credentials"] == "true"
|
|
|
|
|
|
def test_cors_wraps_normal_validation_success_and_preflight_responses(
|
|
operator_client: TestClient,
|
|
) -> None:
|
|
validation = operator_client.post(
|
|
"/api/v1/models",
|
|
headers={"Origin": ALLOWED_ORIGIN, **ADMIN},
|
|
json={},
|
|
)
|
|
assert validation.status_code == 422
|
|
assert validation.headers["access-control-allow-origin"] == ALLOWED_ORIGIN
|
|
|
|
success = operator_client.get(
|
|
"/api/v1/version",
|
|
headers={"Origin": ALLOWED_ORIGIN},
|
|
)
|
|
assert success.status_code == 200
|
|
assert success.headers["access-control-allow-origin"] == ALLOWED_ORIGIN
|
|
|
|
preflight = operator_client.options(
|
|
"/api/v1/models",
|
|
headers={
|
|
"Origin": ALLOWED_ORIGIN,
|
|
"Access-Control-Request-Method": "POST",
|
|
"Access-Control-Request-Headers": "X-ModelForge-Admin-Token,Content-Type",
|
|
},
|
|
)
|
|
assert preflight.status_code == 200
|
|
assert preflight.headers["access-control-allow-origin"] == ALLOWED_ORIGIN
|
|
|
|
non_preflight_options = operator_client.options(
|
|
"/api/v1/agent/future-route-probe",
|
|
headers={"Origin": ALLOWED_ORIGIN},
|
|
)
|
|
assert (
|
|
access_boundary_for_request("OPTIONS", "/api/v1/agent/future-route-probe")
|
|
is AccessBoundary.CONTROL_PLANE
|
|
)
|
|
assert non_preflight_options.status_code == 401
|
|
assert non_preflight_options.headers["access-control-allow-origin"] == ALLOWED_ORIGIN
|
|
|
|
|
|
def test_disallowed_browser_origin_never_receives_allow_origin_header(
|
|
operator_client: TestClient,
|
|
) -> None:
|
|
assert DISALLOWED_ORIGIN not in application_settings.cors_origin_list
|
|
unauthorized = operator_client.post(
|
|
"/api/v1/models",
|
|
headers={"Origin": DISALLOWED_ORIGIN},
|
|
json={},
|
|
)
|
|
assert unauthorized.status_code == 401
|
|
assert "access-control-allow-origin" not in unauthorized.headers
|
|
|
|
preflight = operator_client.options(
|
|
"/api/v1/models",
|
|
headers={
|
|
"Origin": DISALLOWED_ORIGIN,
|
|
"Access-Control-Request-Method": "POST",
|
|
"Access-Control-Request-Headers": "X-ModelForge-Admin-Token",
|
|
},
|
|
)
|
|
assert preflight.status_code == 400
|
|
assert "access-control-allow-origin" not in preflight.headers
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_machine_credential_rejections_consume_zero_body_chunks(
|
|
machine_boundary_harness: MachineBoundaryHarness,
|
|
) -> None:
|
|
node_wrong = f"Bearer mfnode_{uuid.uuid4()}_wrong-prebody-secret"
|
|
cases = [
|
|
("/api/v1/agent/heartbeat", None, 401),
|
|
("/api/v1/agent/heartbeat", node_wrong, 401),
|
|
(
|
|
"/api/v1/agent/heartbeat",
|
|
f"Bearer {machine_boundary_harness.revoked_node_credential}",
|
|
401,
|
|
),
|
|
(
|
|
"/api/v1/agent/heartbeat",
|
|
f"Bearer {machine_boundary_harness.capability_credential}",
|
|
401,
|
|
),
|
|
("/api/v1/capabilities/rag.embedding@1/invoke", None, 401),
|
|
(
|
|
"/api/v1/capabilities/rag.embedding@1/invoke",
|
|
"Bearer mfsvc_wrong-prebody-secret",
|
|
401,
|
|
),
|
|
(
|
|
"/api/v1/capabilities/rag.embedding@1/invoke",
|
|
f"Bearer {machine_boundary_harness.revoked_capability_credential}",
|
|
401,
|
|
),
|
|
(
|
|
"/api/v1/capabilities/rag.embedding@1/invoke",
|
|
f"Bearer {machine_boundary_harness.node_credential}",
|
|
401,
|
|
),
|
|
(
|
|
"/api/v1/capabilities/rag.reranking@1/invoke",
|
|
f"Bearer {machine_boundary_harness.capability_credential}",
|
|
403,
|
|
),
|
|
]
|
|
for path, authorization, expected_status in cases:
|
|
result = await _asgi_request(
|
|
"POST",
|
|
path,
|
|
authorization=authorization,
|
|
body=b"{" + b"x" * (16 * 1024 * 1024),
|
|
fail_on_receive=True,
|
|
)
|
|
assert result.status_code == expected_status
|
|
assert result.body_receive_calls == 0
|
|
assert result.body_bytes_produced == 0
|
|
assert result.headers["access-control-allow-origin"] == ALLOWED_ORIGIN
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_wrong_scope_node_publisher_credential_is_pre_body_403(
|
|
machine_boundary_harness: MachineBoundaryHarness,
|
|
) -> None:
|
|
result = await _asgi_request(
|
|
"POST",
|
|
"/api/v1/agent/heartbeat",
|
|
authorization=f"Bearer {machine_boundary_harness.wrong_scope_node_credential}",
|
|
body=b"{" + b"x" * (16 * 1024 * 1024),
|
|
fail_on_receive=True,
|
|
)
|
|
|
|
assert result.status_code == 403
|
|
assert result.body_receive_calls == 0
|
|
assert json.loads(result.body)["error"]["code"] == "agent_not_authorized"
|
|
assert result.headers["access-control-allow-origin"] == ALLOWED_ORIGIN
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_machine_authentication_backend_unavailability_is_pre_body_503(
|
|
machine_boundary_harness: MachineBoundaryHarness,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
cases = [
|
|
(
|
|
machine_boundary_harness.node_service,
|
|
"/api/v1/agent/heartbeat",
|
|
machine_boundary_harness.node_credential,
|
|
),
|
|
(
|
|
machine_boundary_harness.capability_service,
|
|
"/api/v1/capabilities/rag.embedding@1/invoke",
|
|
machine_boundary_harness.capability_credential,
|
|
),
|
|
]
|
|
for service, path, credential in cases:
|
|
unavailable = Mock(
|
|
side_effect=OperationalError(
|
|
"credential lookup",
|
|
{},
|
|
ConnectionError("authentication database unavailable"),
|
|
)
|
|
)
|
|
with monkeypatch.context() as isolated:
|
|
isolated.setattr(service, "authenticate_with_evidence", unavailable)
|
|
result = await _asgi_request(
|
|
"POST",
|
|
path,
|
|
authorization=f"Bearer {credential}",
|
|
body=b"{" + b"x" * (16 * 1024 * 1024),
|
|
fail_on_receive=True,
|
|
)
|
|
|
|
assert result.status_code == 503
|
|
assert result.body_receive_calls == 0
|
|
assert json.loads(result.body)["error"]["code"] == "machine_authentication_unavailable"
|
|
assert result.headers["access-control-allow-origin"] == ALLOWED_ORIGIN
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_capability_auth_usage_integrity_failure_rolls_back_as_typed_pre_body_503(
|
|
machine_boundary_harness: MachineBoundaryHarness,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
session = machine_boundary_harness.capability_service.session
|
|
failed_commit = Mock(
|
|
side_effect=IntegrityError(
|
|
"UPDATE service_credentials SET last_used_at = ?",
|
|
{},
|
|
RuntimeError("credential usage write failed"),
|
|
)
|
|
)
|
|
rollback = Mock(wraps=session.rollback)
|
|
with monkeypatch.context() as isolated:
|
|
isolated.setattr(session, "commit", failed_commit)
|
|
isolated.setattr(session, "rollback", rollback)
|
|
result = await _asgi_request(
|
|
"POST",
|
|
"/api/v1/capabilities/rag.embedding@1/invoke",
|
|
authorization=f"Bearer {machine_boundary_harness.capability_credential}",
|
|
body=b"{" + b"x" * (16 * 1024 * 1024),
|
|
fail_on_receive=True,
|
|
)
|
|
|
|
assert result.status_code == 503
|
|
assert result.body_receive_calls == 0
|
|
assert json.loads(result.body)["error"]["code"] == "AUTH_BACKEND_UNAVAILABLE"
|
|
assert result.headers["access-control-allow-origin"] == ALLOWED_ORIGIN
|
|
assert failed_commit.call_count == 1
|
|
assert rollback.call_count == 1
|
|
|
|
control = await _asgi_request(
|
|
"POST",
|
|
"/api/v1/capabilities/rag.embedding@1/invoke",
|
|
authorization=f"Bearer {machine_boundary_harness.capability_credential}",
|
|
body=b"{",
|
|
)
|
|
assert control.status_code == 422
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_empty_request_frame_spin_is_bounded_and_cors_wrapped(
|
|
machine_boundary_harness: MachineBoundaryHarness,
|
|
) -> None:
|
|
empty_frames = [
|
|
{"type": "http.request", "body": b"", "more_body": True}
|
|
for _ in range(MAX_CONSECUTIVE_EMPTY_REQUEST_EVENTS + 1)
|
|
]
|
|
result = await _asgi_request(
|
|
"POST",
|
|
"/api/v1/agent/heartbeat",
|
|
authorization=f"Bearer {machine_boundary_harness.node_credential}",
|
|
request_messages=[
|
|
*empty_frames,
|
|
{"type": "http.request", "body": b"{", "more_body": False},
|
|
],
|
|
)
|
|
|
|
assert result.status_code == 400
|
|
assert result.body_receive_calls == MAX_CONSECUTIVE_EMPTY_REQUEST_EVENTS + 1
|
|
assert result.body_bytes_produced == 0
|
|
payload = json.loads(result.body)
|
|
assert payload["error"]["code"] == "request_body_progress_exhausted"
|
|
assert result.headers["x-correlation-id"] == payload["error"]["correlation_id"]
|
|
assert result.headers["access-control-allow-origin"] == ALLOWED_ORIGIN
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_valid_machine_credentials_reach_bounded_body_validation(
|
|
machine_boundary_harness: MachineBoundaryHarness,
|
|
) -> None:
|
|
cases = [
|
|
(
|
|
"/api/v1/agent/heartbeat",
|
|
f"Bearer {machine_boundary_harness.node_credential}",
|
|
),
|
|
(
|
|
"/api/v1/capabilities/rag.embedding@1/invoke",
|
|
f"Bearer {machine_boundary_harness.capability_credential}",
|
|
),
|
|
]
|
|
for path, authorization in cases:
|
|
result = await _asgi_request(
|
|
"POST",
|
|
path,
|
|
authorization=authorization,
|
|
body=b"{",
|
|
)
|
|
assert result.status_code == 422
|
|
assert result.body_receive_calls == 1
|
|
assert result.body_bytes_produced == 1
|
|
assert json.loads(result.body)["error"]["code"] == "request_validation_failed"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_valid_chunked_machine_bodies_stop_at_the_boundary_without_buffering(
|
|
machine_boundary_harness: MachineBoundaryHarness,
|
|
) -> None:
|
|
chunk_bytes = 32_768
|
|
cases = [
|
|
(
|
|
"/api/v1/agent/heartbeat",
|
|
f"Bearer {machine_boundary_harness.node_credential}",
|
|
machine_boundary_harness.settings.node_agent_max_payload_bytes,
|
|
),
|
|
(
|
|
"/api/v1/capabilities/rag.embedding@1/invoke",
|
|
f"Bearer {machine_boundary_harness.capability_credential}",
|
|
machine_boundary_harness.settings.gateway_max_payload_bytes,
|
|
),
|
|
]
|
|
for path, authorization, limit_bytes in cases:
|
|
result = await _asgi_request(
|
|
"POST",
|
|
path,
|
|
authorization=authorization,
|
|
streamed_body_bytes=16 * 1024 * 1024,
|
|
stream_chunk_bytes=chunk_bytes,
|
|
)
|
|
assert result.status_code == 413
|
|
assert json.loads(result.body)["error"] == {
|
|
"code": "request_body_too_large",
|
|
"message": "Request body exceeds the permitted boundary",
|
|
"correlation_id": json.loads(result.body)["error"]["correlation_id"],
|
|
"details": {"limit_bytes": limit_bytes},
|
|
}
|
|
assert limit_bytes < result.body_bytes_produced <= limit_bytes + chunk_bytes
|
|
assert result.body_bytes_produced < 16 * 1024 * 1024
|
|
assert result.headers["access-control-allow-origin"] == ALLOWED_ORIGIN
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_declared_oversize_is_rejected_before_receive_for_all_body_boundaries(
|
|
machine_boundary_harness: MachineBoundaryHarness,
|
|
) -> None:
|
|
cases = [
|
|
(
|
|
"/api/v1/agent/heartbeat",
|
|
f"Bearer {machine_boundary_harness.node_credential}",
|
|
None,
|
|
),
|
|
(
|
|
"/api/v1/capabilities/rag.embedding@1/invoke",
|
|
f"Bearer {machine_boundary_harness.capability_credential}",
|
|
None,
|
|
),
|
|
("/api/v1/agent/enroll", None, None),
|
|
("/api/v1/models", None, ADMIN_TOKEN),
|
|
]
|
|
for path, authorization, operator_token in cases:
|
|
result = await _asgi_request(
|
|
"POST",
|
|
path,
|
|
authorization=authorization,
|
|
operator_token=operator_token,
|
|
content_length=16 * 1024 * 1024,
|
|
fail_on_receive=True,
|
|
)
|
|
assert result.status_code == 413
|
|
assert result.body_receive_calls == 0
|
|
assert result.body_bytes_produced == 0
|
|
assert json.loads(result.body)["error"]["code"] == "request_body_too_large"
|
|
assert result.headers["access-control-allow-origin"] == ALLOWED_ORIGIN
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"path",
|
|
[
|
|
"/api/v1/models",
|
|
"/api/v1/discovery/search",
|
|
"/api/v1/admin/node-enrollments",
|
|
],
|
|
)
|
|
def test_malformed_json_cannot_bypass_pre_body_authentication(
|
|
operator_client: TestClient, path: str
|
|
) -> None:
|
|
missing = operator_client.post(path, content="{", headers={"Content-Type": "application/json"})
|
|
assert missing.status_code == 401
|
|
assert missing.json()["error"]["message"] == "invalid operator credential"
|
|
|
|
invalid = operator_client.post(
|
|
path,
|
|
content="{",
|
|
headers={"Content-Type": "application/json", "X-ModelForge-Admin-Token": "wrong"},
|
|
)
|
|
assert invalid.status_code == 401
|
|
assert invalid.json()["error"]["message"] == "invalid operator credential"
|
|
|
|
legitimate = operator_client.post(
|
|
path,
|
|
content="{",
|
|
headers={"Content-Type": "application/json", **ADMIN},
|
|
)
|
|
assert legitimate.status_code == 422
|
|
assert legitimate.json()["error"]["code"] == "request_validation_failed"
|
|
|
|
|
|
def test_unconfigured_operator_authentication_fails_before_body_parsing() -> None:
|
|
app.dependency_overrides[get_settings] = lambda: Settings(
|
|
_env_file=None,
|
|
operator_api_key=None,
|
|
)
|
|
try:
|
|
response = TestClient(app).post(
|
|
"/api/v1/models",
|
|
content="{",
|
|
headers={"Content-Type": "application/json", "Origin": ALLOWED_ORIGIN},
|
|
)
|
|
finally:
|
|
app.dependency_overrides.clear()
|
|
assert response.status_code == 503
|
|
assert response.json()["error"]["message"] == "operator API authentication is not configured"
|
|
assert response.headers["access-control-allow-origin"] == ALLOWED_ORIGIN
|
|
assert response.headers["access-control-allow-credentials"] == "true"
|
|
|
|
|
|
def test_public_and_machine_boundaries_do_not_accept_or_require_operator_credentials(
|
|
operator_client: TestClient,
|
|
) -> None:
|
|
public = operator_client.get("/api/v1/version")
|
|
assert public.status_code == 200
|
|
|
|
enrollment = operator_client.post(
|
|
"/api/v1/agent/enroll",
|
|
content="{",
|
|
headers={"Content-Type": "application/json"},
|
|
)
|
|
assert enrollment.status_code == 422
|
|
|
|
capability = operator_client.post(
|
|
"/api/v1/capabilities/rag.embedding@1/invoke",
|
|
headers=ADMIN,
|
|
json={"input": ["operator credentials are not capability credentials"]},
|
|
)
|
|
assert capability.status_code == 401
|
|
assert capability.json()["error"]["message"] != "invalid operator credential"
|
|
|
|
node = operator_client.post(
|
|
"/api/v1/agent/heartbeat",
|
|
headers=ADMIN,
|
|
json={},
|
|
)
|
|
assert node.status_code == 401
|
|
assert node.json()["error"]["message"] != "invalid operator credential"
|
|
|
|
|
|
def test_real_scoped_capability_credential_reaches_gateway_but_admin_cannot_substitute() -> None:
|
|
engine = create_engine(
|
|
"sqlite+pysqlite:///:memory:",
|
|
connect_args={"check_same_thread": False},
|
|
poolclass=StaticPool,
|
|
)
|
|
Base.metadata.create_all(engine)
|
|
settings = Settings(_env_file=None, operator_api_key=SecretStr(ADMIN_TOKEN))
|
|
with Session(engine) as session:
|
|
service = ServingService(
|
|
session,
|
|
settings,
|
|
ManifestRegistry(),
|
|
MemoryPayloadStore(),
|
|
)
|
|
service.ensure_contract()
|
|
created = service.create_client(
|
|
ServiceClientCreate(
|
|
name="authorization-boundary-control",
|
|
allowed_capabilities=["rag.embedding@1"],
|
|
)
|
|
)
|
|
app.dependency_overrides[get_settings] = lambda: settings
|
|
app.dependency_overrides[get_serving_service] = lambda: service
|
|
try:
|
|
client = TestClient(app)
|
|
legitimate = client.post(
|
|
"/api/v1/capabilities/rag.embedding@1/invoke",
|
|
headers={"Authorization": f"Bearer {created.credential}"},
|
|
json={"input": "authenticated control"},
|
|
)
|
|
substituted = client.post(
|
|
"/api/v1/capabilities/rag.embedding@1/invoke",
|
|
headers=ADMIN,
|
|
json={"input": "operator credentials are not project credentials"},
|
|
)
|
|
finally:
|
|
app.dependency_overrides.clear()
|
|
|
|
assert legitimate.status_code == 503
|
|
assert legitimate.json()["error"]["code"] == "CAPABILITY_UNAVAILABLE"
|
|
assert substituted.status_code == 401
|
|
assert substituted.json()["error"]["code"] == "CAPABILITY_NOT_AUTHORIZED"
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"confused_credential",
|
|
[
|
|
{"Authorization": "Bearer mfsvc_capability-client-secret"},
|
|
{"Authorization": "Bearer mfnode_node-agent-secret"},
|
|
{"X-ModelForge-Admin-Token": "wrong-operator-secret"},
|
|
],
|
|
)
|
|
def test_capability_node_and_invalid_operator_credentials_cannot_mutate_registry(
|
|
operator_client: TestClient, confused_credential: dict[str, str]
|
|
) -> None:
|
|
response = operator_client.post(
|
|
"/api/v1/models",
|
|
headers=confused_credential,
|
|
json={"attacker_controlled": True},
|
|
)
|
|
assert response.status_code == 401
|
|
assert response.json()["error"]["message"] == "invalid operator credential"
|
|
|
|
|
|
def test_legacy_admin_credential_remains_a_valid_admin_control() -> None:
|
|
engine = create_engine(
|
|
"sqlite+pysqlite:///:memory:",
|
|
connect_args={"check_same_thread": False},
|
|
poolclass=StaticPool,
|
|
)
|
|
Base.metadata.create_all(engine)
|
|
settings = Settings(_env_file=None, operator_api_key=SecretStr(ADMIN_TOKEN))
|
|
with Session(engine) as session:
|
|
app.dependency_overrides[get_settings] = lambda: settings
|
|
app.dependency_overrides[get_registry_service] = lambda: RegistryService(session)
|
|
try:
|
|
response = TestClient(app).post(
|
|
"/api/v1/models",
|
|
headers=ADMIN,
|
|
json={
|
|
"key": "authorized-control",
|
|
"display_name": "Authorized control",
|
|
"source_type": "custom",
|
|
"upstream_provider": "internal",
|
|
"upstream_source": "internal/authorized-control",
|
|
},
|
|
)
|
|
assert response.status_code == 201
|
|
assert response.json()["key"] == "authorized-control"
|
|
finally:
|
|
app.dependency_overrides.clear()
|
|
|
|
|
|
def test_role_hierarchy_is_closed_and_legacy_key_maps_to_admin() -> None:
|
|
viewer = Principal("viewer", PrincipalRole.VIEWER, "test")
|
|
operator = Principal("operator", PrincipalRole.OPERATOR, "test")
|
|
admin = Principal("admin", PrincipalRole.ADMIN, "test")
|
|
|
|
assert require_viewer(viewer) is viewer
|
|
assert require_viewer(operator) is operator
|
|
assert require_operator(operator) is operator
|
|
assert require_admin(admin) is admin
|
|
with pytest.raises(HTTPException) as viewer_escalation:
|
|
require_operator(viewer)
|
|
assert viewer_escalation.value.status_code == 403
|
|
with pytest.raises(HTTPException) as operator_escalation:
|
|
require_admin(operator)
|
|
assert operator_escalation.value.status_code == 403
|
|
|
|
principal = authenticate_operator_token(
|
|
Settings(_env_file=None, operator_api_key=SecretStr(ADMIN_TOKEN)), ADMIN_TOKEN
|
|
)
|
|
assert principal.role is PrincipalRole.ADMIN
|
|
assert principal.authentication_method == "legacy_admin_token"
|
|
|
|
|
|
def test_route_dependency_reuses_the_pre_body_verified_principal() -> None:
|
|
request = Request({"type": "http", "method": "GET", "path": "/", "headers": []})
|
|
principal = Principal("verified-once", PrincipalRole.ADMIN, "legacy_admin_token")
|
|
request.state.principal = principal
|
|
|
|
assert (
|
|
authenticate_operator_principal(
|
|
request,
|
|
Settings(_env_file=None, operator_api_key=None),
|
|
"a value that must not be checked twice",
|
|
)
|
|
is principal
|
|
)
|