361 lines
15 KiB
Python
361 lines
15 KiB
Python
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, client=("127.0.0.1", 8000))
|
|
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_client = TestClient(app, client=("192.0.2.10", 43210))
|
|
remote = remote_client.post("/v1/auth/local-session", json={})
|
|
self.assertEqual(remote.status_code, 403, remote.text)
|
|
|
|
def test_service_rejects_non_loopback_clients_even_for_public_endpoints(self) -> None:
|
|
remote_client = TestClient(app, client=("198.51.100.23", 51000))
|
|
blocked = remote_client.get("/openapi.json")
|
|
local = self.client.get("/openapi.json")
|
|
|
|
self.assertEqual(blocked.status_code, 403, blocked.text)
|
|
self.assertEqual(blocked.json()["error"]["code"], "loopback_only")
|
|
self.assertTrue(blocked.headers.get("x-request-id"))
|
|
self.assertEqual(local.status_code, 200, local.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()
|