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()