167 lines
7.6 KiB
Python
167 lines
7.6 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
from collections.abc import Iterator
|
|
from datetime import UTC, datetime, timedelta
|
|
|
|
import pytest
|
|
from fastapi import HTTPException
|
|
from fastapi.testclient import TestClient
|
|
from pydantic import SecretStr, ValidationError
|
|
from sqlalchemy import create_engine, select
|
|
from sqlalchemy.orm import Session
|
|
from sqlalchemy.pool import StaticPool
|
|
|
|
from modelforge_api.api.authorization import authenticate_operator_token
|
|
from modelforge_api.db import get_session
|
|
from modelforge_api.main import app
|
|
from modelforge_api.persistence.models import AuditEvent, Base
|
|
from modelforge_api.services.startup_validation import validate_settings
|
|
from modelforge_api.settings import Settings, get_settings
|
|
|
|
|
|
def token(role: str) -> str:
|
|
return "mfop_" + role[0] * 43
|
|
|
|
|
|
def record(credential_role: str, **overrides: object) -> dict[str, object]:
|
|
role = credential_role
|
|
return {
|
|
"credential_id": f"test-{role}",
|
|
"subject": f"person-{role}",
|
|
"role": role,
|
|
"token_sha256": hashlib.sha256(token(role).encode()).hexdigest(),
|
|
"expires_at": (datetime.now(UTC) + timedelta(hours=1)).isoformat(),
|
|
**overrides,
|
|
}
|
|
|
|
|
|
def settings(records: list[dict[str, object]] | None = None) -> Settings:
|
|
return Settings(
|
|
_env_file=None, env="test", operator_api_key=None,
|
|
operator_identities=SecretStr(json.dumps(
|
|
records if records is not None else [record(role) for role in ("viewer", "operator", "admin")]
|
|
)),
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def client() -> Iterator[tuple[TestClient, Session]]:
|
|
engine = create_engine("sqlite://", poolclass=StaticPool, connect_args={"check_same_thread": False})
|
|
Base.metadata.create_all(engine)
|
|
with Session(engine) as session:
|
|
app.dependency_overrides[get_settings] = lambda: settings()
|
|
app.dependency_overrides[get_session] = lambda: session
|
|
try:
|
|
yield TestClient(app), session
|
|
finally:
|
|
app.dependency_overrides.clear()
|
|
engine.dispose()
|
|
|
|
|
|
@pytest.mark.parametrize("role", ["viewer", "operator", "admin"])
|
|
def test_real_key_authentication_and_no_store_identity(client: tuple[TestClient, Session], role: str) -> None:
|
|
http, _ = client
|
|
response = http.get("/api/v1/operator/identity", headers={"X-ModelForge-Admin-Token": token(role)})
|
|
assert response.status_code == 200
|
|
assert response.headers["cache-control"] == "no-store"
|
|
assert response.json()["subject"] == f"person-{role}"
|
|
assert response.json()["role"] == role
|
|
assert response.json()["credential_expires_at"] is not None
|
|
assert token(role) not in response.text
|
|
assert "token_sha256" not in response.text
|
|
assert "set-cookie" not in response.headers
|
|
|
|
|
|
def test_roles_and_attributable_configuration_audit(client: tuple[TestClient, Session]) -> None:
|
|
http, database = client
|
|
change = {"expected_revision": 0, "changes": {"gateway_max_batch_size": 2}}
|
|
viewer = {"X-ModelForge-Admin-Token": token("viewer")}
|
|
operator = {"X-ModelForge-Admin-Token": token("operator")}
|
|
admin = {"X-ModelForge-Admin-Token": token("admin"), "X-Correlation-ID": "named-admin-change"}
|
|
assert http.get("/api/v1/configuration/settings", headers=viewer).status_code == 200
|
|
assert http.post("/api/v1/configuration/preview", json=change, headers=viewer).status_code == 403
|
|
preview = http.post("/api/v1/configuration/preview", json=change, headers=operator)
|
|
assert preview.status_code == 200
|
|
body = change | {"preview_digest": preview.json()["preview_digest"], "confirm": True}
|
|
assert http.post("/api/v1/configuration/apply", json=body, headers=operator).status_code == 403
|
|
assert http.post("/api/v1/configuration/apply", json=body, headers=admin).status_code == 200
|
|
event = database.scalar(select(AuditEvent).where(AuditEvent.action == "configuration.applied"))
|
|
assert event is not None and event.actor_id == "person-admin"
|
|
assert event.correlation_id == "named-admin-change"
|
|
assert token("admin") not in str(event.details)
|
|
|
|
|
|
@pytest.mark.parametrize("credential", [None, "unknown", "mfsvc_synthetic", "mfnode_synthetic", "mfenroll_synthetic", "a" * 513, "ünicode"])
|
|
def test_other_credentials_are_refused_before_body(client: tuple[TestClient, Session], credential: str | None) -> None:
|
|
http, _ = client
|
|
response = http.post("/api/v1/configuration/apply", content="not JSON", headers={
|
|
"X-ModelForge-Admin-Token": credential.encode().decode("latin1"),
|
|
} if credential and credential.isascii() else {})
|
|
assert response.status_code == 401
|
|
with pytest.raises(HTTPException) as failure:
|
|
authenticate_operator_token(settings(), credential)
|
|
assert failure.value.status_code == 401
|
|
|
|
|
|
def test_expiry_revocation_rotation_and_no_legacy_fallback() -> None:
|
|
old = record("admin")
|
|
rotated_token = "mfop_" + "R" * 43
|
|
rotated = record("admin", credential_id="rotated", token_sha256=hashlib.sha256(rotated_token.encode()).hexdigest())
|
|
overlap = settings([old, rotated])
|
|
assert authenticate_operator_token(overlap, token("admin")).subject == "person-admin"
|
|
assert authenticate_operator_token(overlap, rotated_token).subject == "person-admin"
|
|
revoked = settings([rotated])
|
|
expired = settings([record("admin", expires_at=datetime.now(UTC).isoformat())])
|
|
for snapshot in (revoked, expired):
|
|
snapshot.operator_api_key = SecretStr(token("admin"))
|
|
with pytest.raises(HTTPException) as failure:
|
|
authenticate_operator_token(snapshot, token("admin"))
|
|
assert failure.value.status_code == 401
|
|
|
|
|
|
@pytest.mark.parametrize("records", [
|
|
[], [record("admin", role="owner")], [record("admin", subject="legacy-operator")],
|
|
[record("admin"), record("admin")],
|
|
[record("viewer", subject="same"), record("admin", subject="same")],
|
|
[record("admin", expires_at="2026-09-09T12:00:00")],
|
|
[record("admin", plaintext="never-reflect-this")],
|
|
])
|
|
def test_invalid_identity_configuration_fails_closed_without_echo(records: list[dict[str, object]]) -> None:
|
|
with pytest.raises(ValidationError) as failure:
|
|
settings(records)
|
|
assert "never-reflect-this" not in str(failure.value)
|
|
assert "token_sha256" not in str(failure.value)
|
|
|
|
|
|
def test_duplicate_json_fields_are_refused() -> None:
|
|
with pytest.raises(ValidationError):
|
|
Settings(_env_file=None, operator_identities=SecretStr('[{"subject":"one","subject":"two"}]'))
|
|
|
|
|
|
def test_malformed_named_namespace_never_falls_back_to_bootstrap() -> None:
|
|
candidate = settings()
|
|
candidate.operator_api_key = SecretStr("mfop_malformed")
|
|
with pytest.raises(HTTPException) as failure:
|
|
authenticate_operator_token(candidate, "mfop_malformed")
|
|
assert failure.value.status_code == 401
|
|
|
|
|
|
def test_named_administrator_can_replace_bootstrap_key_in_production() -> None:
|
|
candidate = settings()
|
|
candidate.env = "production"
|
|
assert not any(issue.setting == "MODELFORGE_OPERATOR_API_KEY" for issue in validate_settings(candidate).problems)
|
|
candidate = settings([record("viewer")])
|
|
candidate.env = "production"
|
|
assert any(issue.setting == "MODELFORGE_OPERATOR_API_KEY" for issue in validate_settings(candidate).problems)
|
|
|
|
|
|
@pytest.mark.parametrize("identities", [None, "", json.dumps([record("admin", expires_at="2020-01-01T00:00:00Z")])])
|
|
def test_production_startup_refuses_missing_or_expired_administration(identities: str | None) -> None:
|
|
candidate = Settings(
|
|
_env_file=None, env="production", operator_api_key=SecretStr(""),
|
|
operator_identities=SecretStr(identities) if identities is not None else None,
|
|
)
|
|
assert any(issue.setting == "MODELFORGE_OPERATOR_API_KEY" for issue in validate_settings(candidate).problems)
|