import json import os import tempfile import unittest from pathlib import Path from unittest.mock import patch from uuid import uuid4 _PROJECT_ROOT = Path(__file__).resolve().parents[1] os.environ.setdefault( "SOVEREIGNAI_DATA_DIR", str(_PROJECT_ROOT / ".test-runtime" / "auth-tests") ) from fastapi.testclient import TestClient from starlette.requests import Request from app import auth, database from app.main import app, create_local_session class AuthenticationTests(unittest.TestCase): def setUp(self) -> None: self.client = TestClient(app) self.created_users: list[str] = [] def tearDown(self) -> None: with database._connect() as connection: connection.executemany( "DELETE FROM users WHERE id=?", [(user_id,) for user_id in self.created_users], ) def _register(self, email: str | None = None) -> tuple[str, str]: account_email = email or f"{uuid4().hex}@example.test" response = self.client.post( "/v1/auth/register", json={"email": account_email, "password": "a long secure passphrase"}, ) self.assertEqual(response.status_code, 201, response.text) data = response.json() self.created_users.append(data["user"]["id"]) return data["user"]["id"], data["access_token"] def test_conversation_api_rejects_missing_or_forged_development_identity(self) -> None: response = self.client.get( "/v1/conversations", headers={"X-User-ID": database.LOCAL_USER_ID}, ) self.assertEqual(response.status_code, 401, response.text) self.assertEqual(response.headers["www-authenticate"], "Bearer") def test_private_api_routes_require_a_valid_bearer_session(self) -> None: for path, method, payload in ( ("/v1/agent/tools", "get", None), ("/v1/agent/skills", "get", None), ("/v1/agent/run", "post", {"task": "hello"}), ("/v1/chat/completions", "post", {"messages": [{"role": "user", "content": "hi"}]}), ("/v1/agent/knowledge/search", "post", {"task": "find", "workspace_path": "."}), ("/v1/web/search", "post", {"query": "test"}), ): response = getattr(self.client, method)(path, json=payload) if payload is not None else getattr(self.client, method)(path) self.assertEqual(response.status_code, 401, f"{method.upper()} {path}: {response.text}") audio = self.client.post( "/v1/audio/transcriptions", files={"file": ("recording.wav", b"", "audio/wav")}, ) self.assertEqual(audio.status_code, 401, audio.text) def test_every_private_v1_openapi_operation_declares_bearer_auth(self) -> None: public = { ("/v1/models", "get"), ("/v1/auth/local-session", "post"), ("/v1/auth/register", "post"), ("/v1/auth/login", "post"), ("/v1/auth/password-reset/request", "post"), ("/v1/auth/password-reset/complete", "post"), } missing: list[str] = [] for path, operations in app.openapi()["paths"].items(): if not path.startswith("/v1/"): continue for method, operation in operations.items(): if (path, method) in public: continue if not operation.get("security"): missing.append(f"{method.upper()} {path}") self.assertEqual(missing, []) def test_agent_audit_history_is_scoped_to_session_owner(self) -> None: _, first_token = self._register() _, second_token = self._register() first_tool = self.client.get( "/v1/agent/tools", headers={"Authorization": f"Bearer {first_token}"} ) second_tool = self.client.get( "/v1/agent/tools", headers={"Authorization": f"Bearer {second_token}"} ) first_audit = self.client.get( "/v1/agent/audit", headers={"Authorization": f"Bearer {first_token}"} ) second_audit = self.client.get( "/v1/agent/audit", headers={"Authorization": f"Bearer {second_token}"} ) self.assertEqual(first_audit.status_code, 200, first_audit.text) self.assertEqual(second_audit.status_code, 200, second_audit.text) self.assertEqual([item["id"] for item in first_audit.json()["data"]], [first_tool.headers["x-agent-audit-id"]]) self.assertEqual([item["id"] for item in second_audit.json()["data"]], [second_tool.headers["x-agent-audit-id"]]) def test_register_login_me_and_logout_use_revocable_sessions(self) -> None: user_id, token = self._register("User@Example.Test") me = self.client.get( "/v1/auth/me", headers={"Authorization": f"Bearer {token}"} ) self.assertEqual(me.status_code, 200, me.text) self.assertEqual(me.json()["user"], { "id": user_id, "email": "user@example.test", "mode": "account", }) with database._connect() as connection: stored = connection.execute( "SELECT password_hash FROM user_identities WHERE user_id=?", (user_id,) ).fetchone() self.assertNotIn("a long secure passphrase", stored["password_hash"]) self.assertEqual( connection.execute( "SELECT COUNT(*) FROM auth_sessions WHERE user_id=?", (user_id,) ).fetchone()[0], 1, ) login = self.client.post( "/v1/auth/login", json={"email": "USER@example.test", "password": "a long secure passphrase"}, ) self.assertEqual(login.status_code, 200, login.text) wrong_password = self.client.post( "/v1/auth/login", json={"email": "user@example.test", "password": "a different wrong phrase"}, ) self.assertEqual(wrong_password.status_code, 401) logout = self.client.post( "/v1/auth/logout", headers={"Authorization": f"Bearer {token}"} ) self.assertEqual(logout.status_code, 200, logout.text) expired = self.client.get( "/v1/auth/me", headers={"Authorization": f"Bearer {token}"} ) self.assertEqual(expired.status_code, 401) def test_password_reset_is_opaque_one_time_and_revokes_existing_sessions(self) -> None: user_id, old_session = self._register("reset@example.test") delivered: list[tuple[str, str | None]] = [] def capture_delivery(email: str, token: str | None) -> None: delivered.append((email, token)) with ( patch("app.main.password_reset_email.smtp_configured", return_value=True), patch( "app.main.password_reset_email.send_password_reset_email", side_effect=capture_delivery, ), ): known = self.client.post( "/v1/auth/password-reset/request", json={"email": "Reset@example.test"} ) unknown = self.client.post( "/v1/auth/password-reset/request", json={"email": "nobody@example.test"} ) self.assertEqual(known.status_code, 202, known.text) self.assertEqual(known.json(), unknown.json()) self.assertEqual(delivered[0][0], "reset@example.test") self.assertIsNotNone(delivered[0][1]) token = delivered[0][1] with database._connect() as connection: stored = connection.execute( "SELECT token_hash FROM password_reset_tokens WHERE user_id=?", (user_id,) ).fetchone() self.assertNotEqual(stored["token_hash"], token) completion = self.client.post( "/v1/auth/password-reset/complete", json={"token": token, "new_password": "a completely new passphrase"}, ) self.assertEqual(completion.status_code, 200, completion.text) old_session_response = self.client.get( "/v1/auth/me", headers={"Authorization": f"Bearer {old_session}"} ) self.assertEqual(old_session_response.status_code, 401) old_password = self.client.post( "/v1/auth/login", json={"email": "reset@example.test", "password": "a long secure passphrase"}, ) new_password = self.client.post( "/v1/auth/login", json={"email": "reset@example.test", "password": "a completely new passphrase"}, ) self.assertEqual(old_password.status_code, 401) self.assertEqual(new_password.status_code, 200, new_password.text) reused = self.client.post( "/v1/auth/password-reset/complete", json={"token": token, "new_password": "another new secure passphrase"}, ) self.assertEqual(reused.status_code, 400) def test_password_reset_tokens_expire(self) -> None: user_id, _ = self._register("expired-reset@example.test") token = auth.issue_password_reset("expired-reset@example.test", now=10_000) self.assertIsNotNone(token) self.assertFalse( auth.reset_password( token or "", "a completely new passphrase", now=10_000 + auth.PASSWORD_RESET_LIFETIME_SECONDS ) ) with database._connect() as connection: count = connection.execute( "SELECT COUNT(*) FROM password_reset_tokens WHERE user_id=?", (user_id,) ).fetchone()[0] self.assertEqual(count, 0) def test_conversations_are_isolated_between_accounts(self) -> None: first_user, first_token = self._register() _, second_token = self._register() conversation_id = str(uuid4()) saved = self.client.put( f"/v1/conversations/{conversation_id}", headers={"Authorization": f"Bearer {first_token}"}, json={ "title": "private", "messages": [{"role": "user", "content": "private message"}], }, ) self.assertEqual(saved.status_code, 200, saved.text) read = self.client.get( f"/v1/conversations/{conversation_id}", headers={"Authorization": f"Bearer {second_token}"}, ) self.assertEqual(read.status_code, 404) first_list = self.client.get( "/v1/conversations", headers={"Authorization": f"Bearer {first_token}"} ) second_list = self.client.get( "/v1/conversations", headers={"Authorization": f"Bearer {second_token}"} ) self.assertEqual(first_list.json()[0]["id"], conversation_id) self.assertEqual(second_list.json(), []) self.assertTrue(first_user) def test_password_account_rejects_duplicate_email(self) -> None: email = f"{uuid4().hex}@example.test" self._register(email) duplicate = self.client.post( "/v1/auth/register", json={"email": email.upper(), "password": "a long secure passphrase"}, ) self.assertEqual(duplicate.status_code, 409, duplicate.text) def test_login_failures_are_rate_limited_and_expire(self) -> None: email = f"{uuid4().hex}@example.test" client_host = f"rate-limit-test-{uuid4().hex}" now = 1_800_000_000 for _ in range(auth.LOGIN_FAILURE_LIMIT): auth.record_login_failure(email, client_host, now=now) self.assertEqual(auth.login_retry_after(email, client_host, now=now), 900) self.assertEqual( auth.login_retry_after(email, client_host, now=now + auth.LOGIN_WINDOW_SECONDS), 0, ) auth.clear_login_failures(email, client_host) def test_workspace_files_are_isolated_by_account_assignment(self) -> None: first_email = f"{uuid4().hex}@example.test" second_email = f"{uuid4().hex}@example.test" _, first_token = self._register(first_email) _, second_token = self._register(second_email) with tempfile.TemporaryDirectory() as directory: parent = Path(directory) first_root = parent / "first" second_root = parent / "second" first_root.mkdir() second_root.mkdir() (first_root / "one.md").write_text("first", encoding="utf-8") (second_root / "two.md").write_text("second", encoding="utf-8") environment = { "SOVEREIGNAI_ALLOWED_WORKSPACES": str(parent), "SOVEREIGNAI_USER_WORKSPACES": json.dumps( { first_email: [str(first_root)], second_email: [str(second_root)], } ), } with patch.dict(os.environ, environment, clear=False): first_own = self.client.post( "/v1/agent/workspace/files", headers={"Authorization": f"Bearer {first_token}"}, json={"workspace_path": str(first_root)}, ) first_other = self.client.post( "/v1/agent/workspace/files", headers={"Authorization": f"Bearer {first_token}"}, json={"workspace_path": str(second_root)}, ) second_own = self.client.post( "/v1/agent/workspace/files", headers={"Authorization": f"Bearer {second_token}"}, json={"workspace_path": str(second_root)}, ) self.assertEqual(first_own.status_code, 200, first_own.text) self.assertEqual(first_own.json()["files"], ["one.md"]) self.assertEqual(first_other.status_code, 403, first_other.text) self.assertEqual(second_own.status_code, 200, second_own.text) self.assertEqual(second_own.json()["files"], ["two.md"]) def test_local_bootstrap_is_restricted_to_loopback_clients(self) -> None: remote = self.client.post("/v1/auth/local-session", json={}) self.assertEqual(remote.status_code, 403, remote.text) request = Request( { "type": "http", "method": "POST", "path": "/v1/auth/local-session", "headers": [], "client": ("127.0.0.1", 43210), "server": ("127.0.0.1", 8000), "scheme": "http", "query_string": b"", } ) token_response = create_local_session(request) self.assertEqual(token_response["user"]["id"], database.LOCAL_USER_ID) authenticated = self.client.get( "/v1/auth/me", headers={"Authorization": f"Bearer {token_response['access_token']}"}, ) self.assertEqual(authenticated.status_code, 200, authenticated.text) if __name__ == "__main__": unittest.main()