312 lines
12 KiB
Python
312 lines
12 KiB
Python
from __future__ import annotations
|
|
|
|
import time
|
|
from unittest.mock import Mock
|
|
|
|
import pytest
|
|
import redis
|
|
from fastapi import HTTPException
|
|
from fastapi.testclient import TestClient
|
|
from pydantic import SecretStr
|
|
|
|
from modelforge_api.api.authorization import authenticate_operator_token
|
|
from modelforge_api.api.routes import operator_login
|
|
from modelforge_api.main import app
|
|
from modelforge_api.services.operator_password import (
|
|
hash_password,
|
|
issue_session,
|
|
parse_password_hash,
|
|
verify_password,
|
|
verify_session,
|
|
)
|
|
from modelforge_api.services.startup_validation import validate_settings
|
|
from modelforge_api.settings import Settings, get_settings
|
|
|
|
PASSWORD = "synthetic password fixture" # noqa: S105 - synthetic test input
|
|
BOOTSTRAP = "synthetic-bootstrap-fixture-" + "x" * 32
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def verifier() -> str:
|
|
return hash_password(PASSWORD)
|
|
|
|
|
|
@pytest.fixture
|
|
def settings(verifier: str) -> Settings:
|
|
return Settings(
|
|
_env_file=None,
|
|
env="test",
|
|
operator_api_key=SecretStr(BOOTSTRAP),
|
|
operator_password_hash=SecretStr(verifier),
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def client(settings: Settings, monkeypatch: pytest.MonkeyPatch):
|
|
connection = Mock()
|
|
connection.__enter__ = Mock(return_value=connection)
|
|
connection.__exit__ = Mock(return_value=False)
|
|
connection.eval.return_value = 1
|
|
monkeypatch.setattr(operator_login.redis.Redis, "from_url", Mock(return_value=connection))
|
|
app.dependency_overrides[get_settings] = lambda: settings
|
|
yield TestClient(app), connection
|
|
app.dependency_overrides.pop(get_settings, None)
|
|
|
|
|
|
def test_salted_password_and_wrong_value(verifier: str):
|
|
assert verify_password(PASSWORD, verifier)
|
|
assert not verify_password("wrong", verifier)
|
|
assert hash_password(PASSWORD) != verifier
|
|
assert len(parse_password_hash(verifier)[0]) == 16
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"encoded",
|
|
[
|
|
"",
|
|
"scrypt$2$8$1$a$b",
|
|
"x" * 10000,
|
|
"scrypt$999999999$8$1$a$b",
|
|
"scrypt$32768$8$1$" + "a" * 66,
|
|
],
|
|
)
|
|
def test_malformed_or_weak_hash_rejected(encoded: str):
|
|
with pytest.raises(ValueError):
|
|
parse_password_hash(encoded)
|
|
configured = Settings(_env_file=None, operator_password_hash=SecretStr(encoded))
|
|
problems = validate_settings(configured).problems
|
|
assert (
|
|
any(p.setting == "MODELFORGE_OPERATOR_PASSWORD_HASH" for p in problems)
|
|
if encoded
|
|
else not any(p.setting == "MODELFORGE_OPERATOR_PASSWORD_HASH" for p in problems)
|
|
)
|
|
assert all(encoded not in p.message for p in problems) if encoded else True
|
|
|
|
|
|
def test_sessions_signed_expiring_rotating(verifier: str):
|
|
token, expiry = issue_session(BOOTSTRAP, verifier, now=1000)
|
|
assert expiry == 1900
|
|
assert verify_session(token, BOOTSTRAP, verifier, now=1000)
|
|
assert verify_session(token, BOOTSTRAP, verifier, now=1899)
|
|
assert not verify_session(token, BOOTSTRAP, verifier, now=1900)
|
|
assert not verify_session(token, BOOTSTRAP, verifier, now=999)
|
|
assert not verify_session(token, BOOTSTRAP + "rotated", verifier, now=1100)
|
|
assert not verify_session(token, BOOTSTRAP, hash_password(PASSWORD), now=1100)
|
|
assert not verify_session(token.replace("1900", "9900"), BOOTSTRAP, verifier, now=1100)
|
|
for malformed in [
|
|
"mfpw_",
|
|
"mfpw_" + "x" * 2000,
|
|
token + ".",
|
|
"mfsvc_" + token,
|
|
"mfpw_1.901.😃.😃",
|
|
]:
|
|
assert not verify_session(malformed, BOOTSTRAP, verifier, now=1100)
|
|
|
|
|
|
def test_login_session_and_legacy_key(client, settings: Settings):
|
|
http, connection = client
|
|
response = http.post("/api/v1/operator/login", json={"password": PASSWORD})
|
|
assert response.status_code == 200
|
|
assert response.headers["cache-control"] == "no-store"
|
|
result = response.json()
|
|
assert result["token_type"] == "bearer" # noqa: S105 - protocol type
|
|
assert PASSWORD not in response.text and BOOTSTRAP not in response.text
|
|
assert result["expires_in"] == 900
|
|
token = result["access_token"]
|
|
principal = authenticate_operator_token(settings, token)
|
|
assert principal.authentication_method == "password_session"
|
|
assert principal.role == "admin"
|
|
assert principal.expires_at is not None
|
|
identity = http.get("/api/v1/operator/identity", headers={"Authorization": f"Bearer {token}"})
|
|
assert identity.status_code == 200
|
|
assert identity.json()["authentication_method"] == "password_session"
|
|
assert identity.json()["credential_expires_at"] == result["expires_at"]
|
|
assert (
|
|
authenticate_operator_token(settings, BOOTSTRAP).authentication_method
|
|
== "legacy_admin_token"
|
|
)
|
|
for headers in [{"Authorization": f"Bearer {token}"}, {"X-ModelForge-Admin-Token": token}]:
|
|
# Routing 404 proves the protected request passed the authentication boundary.
|
|
assert http.get("/api/v1/not-a-route", headers=headers).status_code == 404
|
|
assert (
|
|
http.get("/api/v1/not-a-route", headers={"X-ModelForge-Admin-Token": PASSWORD}).status_code
|
|
== 401
|
|
)
|
|
connection.eval.assert_called_once()
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"payload",
|
|
[
|
|
{},
|
|
{"password": ""},
|
|
{"password": None},
|
|
{"password": 123},
|
|
{"password": ["do-not-echo"]},
|
|
{"password": "z" * 1025},
|
|
{"password": PASSWORD, "do-not-echo": PASSWORD},
|
|
],
|
|
)
|
|
def test_invalid_login_redacted(client, payload):
|
|
http, connection = client
|
|
response = http.post("/api/v1/operator/login", json=payload)
|
|
assert response.status_code == 422
|
|
assert PASSWORD not in response.text and "do-not-echo" not in response.text
|
|
assert response.json()["error"]["details"] == {}
|
|
assert response.headers["cache-control"] == "no-store"
|
|
connection.eval.assert_called_once()
|
|
|
|
|
|
def test_wrong_password_and_no_config(client, settings: Settings):
|
|
http, _ = client
|
|
response = http.post("/api/v1/operator/login", json={"password": "wrong"})
|
|
assert response.status_code == 401 and "wrong" not in response.text
|
|
settings.operator_password_hash = None
|
|
assert http.post("/api/v1/operator/login", json={"password": PASSWORD}).status_code == 503
|
|
|
|
|
|
def test_limiter_runs_before_parser_and_ignores_forwarded_ip(client):
|
|
http, connection = client
|
|
connection.eval.side_effect = [1, 0, 0]
|
|
assert http.post("/api/v1/operator/login", content="not json").status_code == 422
|
|
assert (
|
|
http.post(
|
|
"/api/v1/operator/login",
|
|
json={"password": PASSWORD},
|
|
headers={"X-Forwarded-For": "203.0.113.1"},
|
|
).status_code
|
|
== 429
|
|
)
|
|
assert (
|
|
http.post(
|
|
"/api/v1/operator/login",
|
|
json={"password": PASSWORD},
|
|
headers={"X-Forwarded-For": "203.0.113.2"},
|
|
).status_code
|
|
== 429
|
|
)
|
|
assert len({call.args[-1] for call in connection.eval.call_args_list}) == 1
|
|
script = connection.eval.call_args.args[0]
|
|
assert "redis.call('INCR', KEYS[1])" in script
|
|
assert "redis.call('EXPIRE', KEYS[1], 60)" in script
|
|
assert "global > 30" in script and "peer > 5" in script
|
|
|
|
|
|
def test_redis_failure_fails_closed(client):
|
|
http, connection = client
|
|
connection.eval.side_effect = redis.ConnectionError("secret topology must not leak")
|
|
response = http.post("/api/v1/operator/login", json={"password": PASSWORD})
|
|
assert response.status_code == 503
|
|
assert "topology" not in response.text and PASSWORD not in response.text
|
|
|
|
|
|
def test_invalid_session_never_falls_back(settings: Settings):
|
|
for token in ["mfpw_fake", PASSWORD, None]:
|
|
with pytest.raises(HTTPException) as exc:
|
|
authenticate_operator_token(settings, token)
|
|
assert exc.value.status_code == 401
|
|
token, _ = issue_session(
|
|
BOOTSTRAP, settings.operator_password_hash.get_secret_value(), now=int(time.time()) - 1000
|
|
)
|
|
with pytest.raises(HTTPException):
|
|
authenticate_operator_token(settings, token)
|
|
|
|
|
|
def test_rotations_reject_existing_session(client, settings: Settings):
|
|
http, _ = client
|
|
token = http.post("/api/v1/operator/login", json={"password": PASSWORD}).json()["access_token"]
|
|
headers = {"Authorization": f"Bearer {token}"}
|
|
settings.operator_password_hash = SecretStr(hash_password(PASSWORD))
|
|
assert http.get("/api/v1/not-a-route", headers=headers).status_code == 401
|
|
token = http.post("/api/v1/operator/login", json={"password": PASSWORD}).json()["access_token"]
|
|
settings.operator_api_key = SecretStr(BOOTSTRAP + "rotated")
|
|
assert (
|
|
http.get("/api/v1/not-a-route", headers={"Authorization": f"Bearer {token}"}).status_code
|
|
== 401
|
|
)
|
|
|
|
|
|
def test_login_body_bounded_before_parsing(client):
|
|
http, connection = client
|
|
response = http.post("/api/v1/operator/login", content="x" * 9000)
|
|
assert response.status_code == 413
|
|
assert response.headers["cache-control"] == "no-store"
|
|
assert "x" * 100 not in response.text
|
|
connection.eval.assert_called_once()
|
|
|
|
|
|
def test_login_requires_strong_bootstrap(client, settings: Settings):
|
|
http, _ = client
|
|
settings.operator_api_key = SecretStr("short fixture")
|
|
assert http.post("/api/v1/operator/login", json={"password": PASSWORD}).status_code == 503
|
|
|
|
|
|
def test_cli_stdin_outputs_hash_only():
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
root = Path(__file__).resolve().parents[2]
|
|
result = subprocess.run( # noqa: S603 - fixed local CLI and synthetic stdin
|
|
[sys.executable, str(root / "scripts/operator_password.py")],
|
|
input=PASSWORD + "\n",
|
|
text=True,
|
|
capture_output=True,
|
|
check=False,
|
|
)
|
|
assert result.returncode == 0 and result.stderr == ""
|
|
assert PASSWORD not in result.stdout
|
|
assert verify_password(PASSWORD, result.stdout.strip())
|
|
bad = subprocess.run( # noqa: S603 - fixed local CLI and synthetic stdin
|
|
[sys.executable, str(root / "scripts/operator_password.py")],
|
|
input="z" * 10000,
|
|
text=True,
|
|
capture_output=True,
|
|
check=False,
|
|
)
|
|
assert bad.returncode == 2 and bad.stdout == "" and "z" * 10 not in bad.stderr
|
|
|
|
|
|
def test_empty_optional_password_hash_is_unset():
|
|
for value in ["", SecretStr("")]:
|
|
configured = Settings(_env_file=None, operator_password_hash=value)
|
|
assert configured.operator_password_hash is None
|
|
assert not any(
|
|
p.setting == "MODELFORGE_OPERATOR_PASSWORD_HASH"
|
|
for p in validate_settings(configured).problems
|
|
)
|
|
|
|
|
|
def test_legacy_key_with_session_prefix_remains_accepted(settings: Settings):
|
|
legacy = "mfpw_" + "synthetic-key-fixture" * 3
|
|
settings.operator_api_key = SecretStr(legacy)
|
|
assert (
|
|
authenticate_operator_token(settings, legacy).authentication_method == "legacy_admin_token"
|
|
)
|
|
|
|
|
|
def test_password_configuration_requires_signing_key_in_production(verifier: str):
|
|
configured = Settings(_env_file=None, env="production", operator_api_key=None,
|
|
operator_password_hash=SecretStr(verifier))
|
|
assert any(p.setting == "MODELFORGE_OPERATOR_API_KEY" and "session signing" in p.message
|
|
for p in validate_settings(configured).problems)
|
|
|
|
|
|
@pytest.mark.parametrize("bootstrap", [None, "", "short", "x" * 31, "x" * 32])
|
|
def test_named_admin_cannot_replace_password_signing_key(verifier: str, bootstrap: str | None):
|
|
import hashlib
|
|
import json
|
|
from datetime import UTC, datetime, timedelta
|
|
|
|
identities = SecretStr(json.dumps([{
|
|
"credential_id": "fixture-admin", "subject": "fixture-person", "role": "admin",
|
|
"token_sha256": hashlib.sha256(("mfop_" + "a" * 43).encode()).hexdigest(),
|
|
"expires_at": (datetime.now(UTC) + timedelta(hours=1)).isoformat(),
|
|
}]))
|
|
configured = Settings(_env_file=None, env="production", operator_identities=identities,
|
|
operator_api_key=None if bootstrap is None else SecretStr(bootstrap),
|
|
operator_password_hash=SecretStr(verifier))
|
|
signing_problems = [p for p in validate_settings(configured).problems
|
|
if p.setting == "MODELFORGE_OPERATOR_API_KEY" and "session signing" in p.message]
|
|
assert bool(signing_problems) == (bootstrap is None or len(bootstrap) < 32)
|