feat(cloud-console): add user authentication and administration
This commit is contained in:
@@ -0,0 +1,258 @@
|
||||
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()
|
||||
Reference in New Issue
Block a user