259 lines
9.3 KiB
Python
259 lines
9.3 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import timedelta
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
from sqlalchemy import text
|
|
|
|
from cloud.control_config import CloudControlConfig
|
|
from cloud.database import CloudDatabase
|
|
from cloud.repository import LastAdministratorConflictError
|
|
from cloud.sdk.client import CloudAuthorizationError, CloudClient
|
|
from cloud.user_auth import UserAuthService, UserAuthSettings, UserAuthenticationError, utc_now
|
|
from cloud_api.app import create_app
|
|
|
|
|
|
def _create_admin(client: TestClient):
|
|
service = client.app.state.cloud_services.user_auth_service
|
|
return service.create_user(
|
|
username="admin",
|
|
display_name="Administrator",
|
|
role="admin",
|
|
password="correct-horse-battery-staple",
|
|
must_change_password=False,
|
|
)
|
|
|
|
|
|
def test_user_login_session_csrf_and_admin_lifecycle() -> None:
|
|
app = create_app(config=CloudControlConfig(database_url="sqlite:///:memory:"))
|
|
with TestClient(app) as client:
|
|
admin = _create_admin(client)
|
|
login = client.post(
|
|
"/v1/auth/login",
|
|
json={"username": "ADMIN", "password": "correct-horse-battery-staple"},
|
|
)
|
|
|
|
assert login.status_code == 200
|
|
assert login.json()["role"] == "admin"
|
|
assert "amcp_session" in login.headers["set-cookie"]
|
|
assert "HttpOnly" in login.headers["set-cookie"]
|
|
assert client.get("/v1/auth/me").json()["id"] == admin.id
|
|
|
|
missing_csrf = client.post(
|
|
"/v1/users",
|
|
json={
|
|
"username": "viewer",
|
|
"display_name": "Viewer",
|
|
"role": "viewer",
|
|
"password": "another-secure-password",
|
|
},
|
|
)
|
|
assert missing_csrf.status_code == 403
|
|
|
|
csrf = client.cookies.get("amcp_csrf")
|
|
created = client.post(
|
|
"/v1/users",
|
|
headers={"X-CSRF-Token": csrf},
|
|
json={
|
|
"username": "viewer",
|
|
"display_name": "Viewer",
|
|
"role": "viewer",
|
|
"password": "another-secure-password",
|
|
},
|
|
)
|
|
assert created.status_code == 201
|
|
viewer_id = created.json()["id"]
|
|
assert created.json()["must_change_password"] is True
|
|
|
|
users = client.get("/v1/users")
|
|
assert users.status_code == 200
|
|
assert {item["id"] for item in users.json()["items"]} == {admin.id, viewer_id}
|
|
|
|
logout = client.post("/v1/auth/logout", headers={"X-CSRF-Token": csrf})
|
|
assert logout.status_code == 204
|
|
assert client.get("/v1/auth/me").status_code == 401
|
|
|
|
|
|
def test_session_user_is_scope_limited_and_must_change_password() -> None:
|
|
app = create_app(config=CloudControlConfig(database_url="sqlite:///:memory:"))
|
|
with TestClient(app) as client:
|
|
service = client.app.state.cloud_services.user_auth_service
|
|
service.create_user(
|
|
username="operator",
|
|
display_name="Operator",
|
|
role="operator",
|
|
password="correct-horse-battery-staple",
|
|
must_change_password=False,
|
|
)
|
|
service.create_user(
|
|
username="temporary",
|
|
display_name="Temporary",
|
|
role="viewer",
|
|
password="correct-horse-battery-staple",
|
|
)
|
|
|
|
assert client.post(
|
|
"/v1/auth/login",
|
|
json={"username": "operator", "password": "correct-horse-battery-staple"},
|
|
).status_code == 200
|
|
csrf = client.cookies.get("amcp_csrf")
|
|
assert client.get("/v1/tasks").status_code == 200
|
|
assert client.post(
|
|
"/v1/tasks",
|
|
headers={"X-CSRF-Token": csrf},
|
|
json={"goal": "inspect"},
|
|
).status_code == 201
|
|
assert client.get("/v1/users").status_code == 403
|
|
client.post("/v1/auth/logout", headers={"X-CSRF-Token": csrf})
|
|
|
|
assert client.post(
|
|
"/v1/auth/login",
|
|
json={"username": "temporary", "password": "correct-horse-battery-staple"},
|
|
).status_code == 200
|
|
assert client.get("/v1/tasks").status_code == 403
|
|
assert client.get("/v1/auth/me").status_code == 200
|
|
|
|
|
|
def test_login_failure_is_generic_and_throttled() -> None:
|
|
app = create_app(
|
|
config=CloudControlConfig(
|
|
database_url="sqlite:///:memory:",
|
|
login_failure_limit=2,
|
|
login_block_seconds=60,
|
|
)
|
|
)
|
|
with TestClient(app) as client:
|
|
_create_admin(client)
|
|
for password in ("wrong-password", "wrong-password", "correct-horse-battery-staple"):
|
|
response = client.post(
|
|
"/v1/auth/login",
|
|
json={"username": "admin", "password": password},
|
|
)
|
|
assert response.status_code == 401
|
|
assert response.json()["detail"] == "invalid username or password"
|
|
|
|
|
|
def test_password_change_revokes_existing_session() -> None:
|
|
app = create_app(config=CloudControlConfig(database_url="sqlite:///:memory:"))
|
|
with TestClient(app) as client:
|
|
admin = _create_admin(client)
|
|
client.post(
|
|
"/v1/auth/login",
|
|
json={"username": "admin", "password": "correct-horse-battery-staple"},
|
|
)
|
|
csrf = client.cookies.get("amcp_csrf")
|
|
changed = client.post(
|
|
"/v1/auth/password",
|
|
headers={"X-CSRF-Token": csrf},
|
|
json={
|
|
"current_password": "correct-horse-battery-staple",
|
|
"new_password": "new-correct-horse-battery-staple",
|
|
},
|
|
)
|
|
assert changed.status_code == 204
|
|
assert client.get("/v1/auth/me").status_code == 401
|
|
relogin = client.post(
|
|
"/v1/auth/login",
|
|
json={"username": "admin", "password": "new-correct-horse-battery-staple"},
|
|
)
|
|
assert relogin.status_code == 200
|
|
assert relogin.json()["id"] == admin.id
|
|
|
|
|
|
def test_last_administrator_is_preserved() -> None:
|
|
app = create_app(config=CloudControlConfig(database_url="sqlite:///:memory:"))
|
|
with TestClient(app) as client:
|
|
admin = _create_admin(client)
|
|
repository = client.app.state.cloud_services.repository
|
|
with pytest.raises(LastAdministratorConflictError):
|
|
repository.update_user(
|
|
admin.id,
|
|
enabled=False,
|
|
updated_at=utc_now(),
|
|
)
|
|
|
|
|
|
def test_user_auth_service_expires_sessions() -> None:
|
|
app = create_app(config=CloudControlConfig(database_url="sqlite:///:memory:"))
|
|
with TestClient(app) as client:
|
|
repository = client.app.state.cloud_services.repository
|
|
service = UserAuthService(
|
|
repository,
|
|
settings=UserAuthSettings(
|
|
session_idle_ttl=timedelta(seconds=1),
|
|
session_absolute_ttl=timedelta(seconds=1),
|
|
),
|
|
)
|
|
service.create_user(
|
|
username="expired",
|
|
display_name="Expired",
|
|
role="viewer",
|
|
password="correct-horse-battery-staple",
|
|
must_change_password=False,
|
|
)
|
|
login = service.login(
|
|
username="expired",
|
|
password="correct-horse-battery-staple",
|
|
client_bucket="test",
|
|
now=utc_now(),
|
|
)
|
|
assert service.authenticate_session(
|
|
login.session_token,
|
|
now=utc_now() + timedelta(seconds=2),
|
|
) is None
|
|
|
|
|
|
def test_cloud_client_preserves_user_session_and_csrf() -> None:
|
|
app = create_app(config=CloudControlConfig(database_url="sqlite:///:memory:"))
|
|
with TestClient(app) as http_client:
|
|
_create_admin(http_client)
|
|
client = CloudClient("http://testserver", http_client=http_client)
|
|
assert client.login(
|
|
username="admin",
|
|
password="correct-horse-battery-staple",
|
|
)["username"] == "admin"
|
|
assert client.current_user()["role"] == "admin"
|
|
created = client.create_user(
|
|
username="client-user",
|
|
display_name="Client User",
|
|
role="viewer",
|
|
password="another-secure-password",
|
|
)
|
|
assert created["must_change_password"] is True
|
|
assert {item["username"] for item in client.list_users()["items"]} == {
|
|
"admin",
|
|
"client-user",
|
|
}
|
|
client.logout()
|
|
with pytest.raises(CloudAuthorizationError):
|
|
client.current_user()
|
|
|
|
|
|
def test_audit_events_do_not_contain_password_or_session_secrets() -> None:
|
|
database = CloudDatabase("sqlite:///:memory:")
|
|
try:
|
|
service = UserAuthService(database.repository, settings=UserAuthSettings())
|
|
service.create_user(
|
|
username="admin",
|
|
display_name="Administrator",
|
|
role="admin",
|
|
password="correct-horse-battery-staple",
|
|
must_change_password=False,
|
|
)
|
|
with pytest.raises(UserAuthenticationError):
|
|
service.login(
|
|
username="admin",
|
|
password="incorrect-secret-password",
|
|
client_bucket="127.0.0.1",
|
|
)
|
|
with database.engine.connect() as connection:
|
|
values = connection.scalars(
|
|
text("select metadata_json from cloud_auth_audit_events")
|
|
).all()
|
|
rendered = " ".join(values)
|
|
assert "incorrect-secret-password" not in rendered
|
|
assert "correct-horse-battery-staple" not in rendered
|
|
finally:
|
|
database.close()
|