fischer
/
fischer-agentkit
1
0
Fork 0
Code Issues Pull Requests Actions 87 Packages Projects Releases Wiki Activity

feat(auth): U3/U4/U9 logout-others + whoami cold-start + admin UI + integration tests

This commit is contained in:
chiguyong 2026-06-21 09:08:34 +08:00
parent 9328451050
commit aee7362665
9 changed files with 1142 additions and 41 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
src/agentkit/server/auth/middleware.py
View File

@ -47,6 +47,7 @@ class AuthMiddleware(BaseHTTPMiddleware):
"/api/v1/auth/login",
"/api/v1/auth/refresh",
"/api/v1/auth/logout",
"/api/v1/auth/whoami", # Route does its own auth (access OR refresh)
"/docs",
"/openapi.json",
"/redoc",
@ -70,8 +71,21 @@ class AuthMiddleware(BaseHTTPMiddleware):
# ------------------------------------------------------------------
def _is_whitelisted(self, path: str) -> bool:
"""Return True if ``path`` starts with any whitelisted prefix."""
return any(path.startswith(p) for p in self.WHITELIST_PATHS)
"""Return True if ``path`` matches a whitelisted route.
Uses exact match for auth routes (so ``/auth/logout`` does NOT
whitelist ``/auth/logout-others``) and prefix match for docs.
"""
for prefix in self.WHITELIST_PATHS:
if path == prefix:
return True
# Prefix match only for documentation paths (trailing slash
# or sub-path is fine). Auth paths require exact match to
# avoid accidentally whitelisting sibling routes like
# /auth/logout-others under /auth/logout.
if prefix in ("/docs", "/openapi.json", "/redoc") and path.startswith(prefix):
return True
return False
def _is_dev_mode(self) -> bool:
"""Dev mode = no JWT secret, no global API key, no client keys."""

43
src/agentkit/server/auth/session_service.py
View File

@ -203,6 +203,23 @@ class SessionService:
rows = await cursor.fetchall()
return [_row_to_info(r) for r in rows]
async def list_active_by_provider(self, auth_provider: str) -> list[SessionInfo]:
"""List active sessions for a given auth provider (KTD-10).
Supports the future "show me all OIDC sessions" admin view.
Only non-revoked, non-expired sessions are returned.
"""
sql = (
"SELECT * FROM auth_sessions "
"WHERE auth_provider = ? AND revoked = 0 "
"ORDER BY created_at DESC"
)
async with aiosqlite.connect(str(self._db_path)) as db:
db.row_factory = aiosqlite.Row
cursor = await db.execute(sql, (auth_provider,))
rows = await cursor.fetchall()
return [_row_to_info(r) for r in rows]
async def find_by_refresh_token(self, refresh_token: str) -> SessionInfo | None:
"""Look up a session by the SHA-256 hash of its refresh token."""
h = hash_token(refresh_token)
@ -378,7 +395,11 @@ class SessionService:
return await self.revoke(info.id, reason=reason)
async def revoke_all_for_user(
self, user_id: str, *, reason: str = REVOKE_REASON_USER_TERMINATED
self,
user_id: str,
*,
except_sid: str | None = None,
reason: str = REVOKE_REASON_USER_TERMINATED,
) -> int:
"""Revoke all of a user's active sessions.
@ -386,14 +407,22 @@ class SessionService:
- The user changes their password (invalidate all other devices).
- Token reuse is detected (invalidate everything as a precaution).
- An admin kills a user account.
- The user calls ``/auth/logout-others`` (keep current session).
If ``except_sid`` is provided, that session is spared (e.g. the
caller's current session on "logout others").
"""
sql = (
"UPDATE auth_sessions "
"SET revoked = 1, revoked_reason = ? "
"WHERE user_id = ? AND revoked = 0"
)
args: list[Any] = [reason, user_id]
if except_sid is not None:
sql += " AND id != ?"
args.append(except_sid)
async with aiosqlite.connect(str(self._db_path)) as db:
cursor = await db.execute(
"UPDATE auth_sessions "
"SET revoked = 1, revoked_reason = ? "
"WHERE user_id = ? AND revoked = 0",
(reason, user_id),
)
cursor = await db.execute(sql, tuple(args))
await db.commit()
return cursor.rowcount

15
src/agentkit/server/frontend/src/components/layout/TopNav.vue
View File

@ -37,6 +37,11 @@
<SettingOutlined />
</button>
</a-tooltip>
<a-tooltip v-if="authStore.isAdmin()" title="用户与会话管理">
<button class="top-nav__icon-btn" @click="router.push('/admin/users')">
<TeamOutlined />
</button>
</a-tooltip>
</div>
</header>
</template>
@ -45,9 +50,16 @@
import { computed } from 'vue'
import { useRouter } from 'vue-router'
import { Badge as ABadge, Tooltip as ATooltip } from 'ant-design-vue'
import { SettingOutlined, MenuFoldOutlined, MenuUnfoldOutlined, BulbOutlined } from '@ant-design/icons-vue'
import {
SettingOutlined,
MenuFoldOutlined,
MenuUnfoldOutlined,
BulbOutlined,
TeamOutlined,
} from '@ant-design/icons-vue'
import { useChatStore } from '@/stores/chat'
import { useThemeStore } from '@/stores/theme'
import { useAuthStore } from '@/stores/auth'
const props = defineProps<{
iconNavCollapsed?: boolean
@ -60,6 +72,7 @@ const emit = defineEmits<{
const router = useRouter()
const chatStore = useChatStore()
const themeStore = useThemeStore()
const authStore = useAuthStore()
const wsConnected = computed(() => chatStore.isWsConnected)
const iconNavCollapsed = computed(() => props.iconNavCollapsed ?? false)
const isDark = computed(() => themeStore.resolvedMode === 'dark')

28
src/agentkit/server/frontend/src/router/index.ts
View File

@ -89,6 +89,14 @@ const routes: RouteRecordRaw[] = [
meta: { title: 'Computer Use' },
},
// Admin: user sessions management (U9)
{
path: '/admin/users',
name: 'admin-users',
component: () => import('@/views/admin/UsersView.vue'),
meta: { title: '用户与会话管理', requiresAdmin: true },
},
// Legacy layout (fallback)
{
path: '/legacy',
@ -186,16 +194,22 @@ router.beforeEach((to, _from, next) => {
}
const authStore = useAuthStore()
if (authStore.isAuthenticated) {
next()
if (!authStore.isAuthenticated) {
// Preserve the original target so we can redirect after login
next({
name: 'login',
query: { redirect: to.fullPath },
})
return
}
// Preserve the original target so we can redirect after login
next({
name: 'login',
query: { redirect: to.fullPath },
})
// Admin-only routes require the admin role.
if (to.meta.requiresAdmin === true && !authStore.isAdmin()) {
next({ name: 'agent-chat' })
return
}
next()
})
export default router

219
src/agentkit/server/frontend/src/views/admin/UsersView.vue Normal file
View File

@ -0,0 +1,219 @@
<template>
<div class="users-view">
<a-page-header
title="用户与会话管理"
sub-title="管理员可查看和撤销任意用户的会话"
class="users-view__header"
/>
<a-tabs v-model:activeKey="activeTab" class="users-view__tabs">
<!-- Tab 1: 按用户查询会话 -->
<a-tab-pane key="by-user" tab="按用户查询">
<a-card class="users-view__card">
<a-space direction="vertical" :size="16" style="width: 100%">
<a-form layout="inline">
<a-form-item label="用户 ID">
<a-input
v-model:value="userIdInput"
placeholder="输入用户 ID（UUID）"
style="width: 360px"
allow-clear
@press-enter="loadUserSessions"
/>
</a-form-item>
<a-form-item>
<a-button
type="primary"
:loading="loadingUser"
:disabled="!userIdInput.trim()"
@click="loadUserSessions"
>
查询会话
</a-button>
</a-form-item>
</a-form>
<a-alert
v-if="userQueryError"
type="error"
:message="userQueryError"
show-icon
closable
@close="userQueryError = ''"
/>
<div v-if="selectedUserId" class="users-view__sessions">
<a-divider orientation="left">
用户 <code>{{ selectedUserId }}</code> 的会话
</a-divider>
<UserSessionsPanel :user-id="selectedUserId" />
</div>
<a-empty
v-else
description="输入用户 ID 后点击查询，即可查看该用户的所有会话"
/>
</a-space>
</a-card>
</a-tab-pane>
<!-- Tab 2: 全局会话概览 -->
<a-tab-pane key="all-sessions" tab="全局会话概览">
<a-card class="users-view__card">
<a-space direction="vertical" :size="16" style="width: 100%">
<a-space :size="8" align="center">
<a-button :loading="loadingAll" @click="loadAllSessions">刷新</a-button>
<span class="users-view__count">共 {{ allSessions.length }} 条会话</span>
</a-space>
<a-alert
v-if="allSessionsError"
type="error"
:message="allSessionsError"
show-icon
closable
@close="allSessionsError = ''"
/>
<ActiveSessionsPanel
:sessions="allSessions"
:loading="loadingAll"
:revoking-id="revokingAllId"
:admin-mode="true"
@refresh="loadAllSessions"
@revoke="handleRevokeAll"
/>
</a-space>
</a-card>
</a-tab-pane>
</a-tabs>
</div>
</template>
<script setup lang="ts">
/**
* Admin: user sessions management view (U9).
*
* Two tabs:
* 1. "按用户查询" — enter a user ID, see that user's sessions via
* :class:`UserSessionsPanel` (which wraps the shared
* :class:`ActiveSessionsPanel` in admin mode).
* 2. "全局会话概览" — list all recent sessions across the system,
* with the ability to revoke any of them.
*
* Access to this view is gated by the router (admin role required).
*/
import { ref, onMounted } from 'vue'
import { message } from 'ant-design-vue'
import UserSessionsPanel from '@/components/admin/UserSessionsPanel.vue'
import ActiveSessionsPanel from '@/components/settings/ActiveSessionsPanel.vue'
import { adminApi } from '@/api/admin'
import type { ISessionInfo } from '@/api/auth'
const activeTab = ref<'by-user' | 'all-sessions'>('by-user')
// --- Tab 1: by user ---
const userIdInput = ref('')
const selectedUserId = ref('')
const loadingUser = ref(false)
const userQueryError = ref('')
async function loadUserSessions(): Promise<void> {
const uid = userIdInput.value.trim()
if (!uid) return
loadingUser.value = true
userQueryError.value = ''
try {
// Just verify the user exists by listing their sessions; the panel
// does its own data loading once we set selectedUserId.
await adminApi.listUserSessions(uid, false)
selectedUserId.value = uid
} catch (err) {
userQueryError.value = _extractMessage(err, '查询用户会话失败（请检查用户 ID 是否正确）')
selectedUserId.value = ''
} finally {
loadingUser.value = false
}
}
// --- Tab 2: all sessions ---
const allSessions = ref<ISessionInfo[]>([])
const loadingAll = ref(false)
const allSessionsError = ref('')
const revokingAllId = ref<string | null>(null)
async function loadAllSessions(): Promise<void> {
loadingAll.value = true
allSessionsError.value = ''
try {
allSessions.value = await adminApi.listAllSessions(200)
} catch (err) {
allSessionsError.value = _extractMessage(err, '加载全局会话列表失败')
} finally {
loadingAll.value = false
}
}
async function handleRevokeAll(sid: string): Promise<void> {
// The global overview doesn't have a userId per row in the current
// ISessionInfo shape — but the server's DELETE /admin/sessions/{sid}
// endpoint doesn't require a userId. We use the per-user endpoint
// only when we know the userId. For the global list, find the row.
const row = allSessions.value.find((s) => s.id === sid)
if (!row || !row.user_id) {
message.error('无法撤销：缺少用户 ID 信息')
return
}
revokingAllId.value = sid
try {
const result = await adminApi.revokeUserSession(row.user_id, sid)
if (result.revoked) {
message.success('已撤销该会话')
} else {
message.info('该会话已是撤销状态')
}
allSessions.value = allSessions.value.filter((s) => s.id !== sid)
} catch (err) {
message.error(_extractMessage(err, '撤销会话失败'))
} finally {
revokingAllId.value = null
}
}
function _extractMessage(err: unknown, fallback: string): string {
if (err && typeof err === 'object') {
const obj = err as { detail?: unknown; message?: unknown }
if (typeof obj.detail === 'string' && obj.detail) return obj.detail
if (typeof obj.message === 'string' && obj.message) return obj.message
}
return fallback
}
onMounted(() => {
loadAllSessions()
})
</script>
<style scoped>
.users-view {
padding: 24px;
max-width: 1200px;
margin: 0 auto;
}
.users-view__header {
margin-bottom: 16px;
}
.users-view__card {
margin-top: 8px;
}
.users-view__sessions {
margin-top: 8px;
}
.users-view__count {
color: var(--text-secondary, #666);
font-size: 14px;
}
</style>

154
src/agentkit/server/routes/auth.py
View File

@ -27,6 +27,7 @@ from pathlib import Path
from typing import Any
import aiosqlite
import jwt
from fastapi import APIRouter, Depends, HTTPException, Request
from pydantic import BaseModel, ConfigDict, EmailStr
@ -137,6 +138,7 @@ class SessionResponse(BaseModel):
revoked: bool = False
revoked_reason: str | None = None
user_id: str | None = None
previous_session_id: str | None = None
class TokenResponse(BaseModel):
@ -152,6 +154,23 @@ class TokenResponse(BaseModel):
session_id: str
class WhoamiResponse(BaseModel):
"""Response of ``GET /auth/whoami`` (cold-start + session metadata).
When the client calls whoami with a refresh token (cold-start), the
server issues a fresh access token so the client doesn't need a
separate ``/auth/refresh`` round-trip. When called with an access
token, ``access_token`` is ``None`` (the caller already has one).
"""
model_config = ConfigDict(extra="forbid")
user: UserResponse
access_token: str | None = None
session_id: str | None = None
session: SessionResponse | None = None
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
@ -514,21 +533,42 @@ async def logout(payload: LogoutRequest, request: Request) -> dict[str, Any]:
return {"revoked": revoked}
@router.get("/whoami", response_model=UserResponse)
async def whoami(
request: Request,
user: dict[str, Any] = Depends(require_authenticated),
) -> UserResponse:
"""Return the current authenticated user's profile.
@router.get("/whoami", response_model=WhoamiResponse)
async def whoami(request: Request) -> WhoamiResponse:
"""Return the current user + session metadata (cold-start support).
Used by the client for cold-start (after restoring a refresh token
from local storage) to verify the session is still valid AND fetch
the freshest user profile.
Accepts **either** an access token (normal call) **or** a refresh
token (cold-start, when the access token is gone). The route does
its own auth because the middleware only accepts access tokens.
On cold-start (refresh token presented), the server issues a fresh
access token so the client doesn't need a separate ``/auth/refresh``
round-trip. On 401 from this endpoint, the client treats it as
'invalid' state (NOT 'error' state) so the router redirects to
/login.
"""
user_id = user.get("user_id")
if not user_id:
raise HTTPException(status_code=401, detail="Authentication required")
auth_header = request.headers.get("Authorization", "")
if not auth_header.startswith("Bearer "):
raise HTTPException(status_code=401, detail="missing bearer token")
token = auth_header[7:]
secret = _resolve_jwt_secret(request)
try:
# Accept both access and refresh tokens for cold-start.
payload = verify_token(token, secret, expected_type=None)
except jwt.ExpiredSignatureError:
raise HTTPException(status_code=401, detail="token expired")
except jwt.InvalidTokenError:
raise HTTPException(status_code=401, detail="invalid token")
user_id = payload.get("sub")
if not user_id:
raise HTTPException(status_code=401, detail="invalid token: no subject")
token_type = payload.get("type")
sid = payload.get("sid")
# Load the user row.
db_path = await _ensure_db(request)
async with aiosqlite.connect(str(db_path)) as db:
db.row_factory = aiosqlite.Row
@ -536,7 +576,38 @@ async def whoami(
row = await cursor.fetchone()
if row is None:
raise HTTPException(status_code=404, detail="User not found")
return _user_row_to_response(row)
user_response = _user_row_to_response(row)
# V2 token with sid: validate the session is still active.
session_response: SessionResponse | None = None
if sid:
svc: SessionService = get_session_service()
info = await svc.get(sid)
if info is None or info.revoked:
raise HTTPException(status_code=401, detail="session revoked or expired")
session_response = SessionResponse(
**auth_session_row_to_dict(_info_to_dict(info)),
is_current=True,
)
# Cold-start: refresh token presented → issue a fresh access token.
new_access_token: str | None = None
if token_type == "refresh":
new_pair = create_token_pair(
user_id=str(row["id"]),
username=str(row["username"]),
role=str(row["role"]),
secret=secret,
session_id=sid,
)
new_access_token = new_pair.access_token
return WhoamiResponse(
user=user_response,
access_token=new_access_token,
session_id=sid,
session=session_response,
)
@router.get("/sessions", response_model=list[SessionResponse])
@ -558,9 +629,6 @@ async def list_sessions(
_info_to_dict(s),
),
is_current=(s.id == current_sid),
revoked=s.revoked,
revoked_reason=s.revoked_reason,
user_id=s.user_id,
)
for s in sessions
]
@ -604,6 +672,35 @@ async def revoke_own_session(
return {"revoked": ok}
@router.post("/logout-others")
async def logout_others(
request: Request,
user: dict[str, Any] = Depends(require_authenticated),
) -> dict[str, Any]:
"""Revoke all of the current user's sessions except the calling one.
Used by the "Log out other devices" button in the security settings.
The current session (identified by the JWT's ``sid`` claim) is spared.
"""
user_id = user.get("user_id")
if not user_id:
raise HTTPException(status_code=401, detail="Authentication required")
current_sid = user.get("sid")
if not current_sid:
# Legacy token without sid — can't identify the current session.
raise HTTPException(
status_code=400,
detail="Current session cannot be identified (legacy token). Please log in again.",
)
svc: SessionService = get_session_service()
count = await svc.revoke_all_for_user(
user_id,
except_sid=current_sid,
reason=REVOKE_REASON_USER_TERMINATED,
)
return {"revoked_count": count}
@router.post("/change-password")
async def change_password(
payload: ChangePasswordRequest,
@ -677,9 +774,6 @@ async def admin_list_sessions(
SessionResponse(
**auth_session_row_to_dict(_info_to_dict(s)),
is_current=False,
revoked=s.revoked,
revoked_reason=s.revoked_reason,
user_id=s.user_id,
)
for s in sessions
]
@ -734,9 +828,6 @@ async def admin_list_user_sessions(
SessionResponse(
**auth_session_row_to_dict(_info_to_dict(s)),
is_current=(s.id == current_sid),
revoked=s.revoked,
revoked_reason=s.revoked_reason,
user_id=s.user_id,
)
for s in sessions
]
@ -770,5 +861,20 @@ async def me(
request: Request,
user: dict[str, Any] = Depends(require_authenticated),
) -> UserResponse:
"""Alias for :func:`whoami` (legacy path)."""
return await whoami(request=request, user=user)
"""Alias for :func:`whoami` (legacy path, access-token only).
Returns just the user profile (no session metadata) for backwards
compatibility with clients that expect the old ``UserResponse``
shape from ``/auth/me``.
"""
user_id = user.get("user_id")
if not user_id:
raise HTTPException(status_code=401, detail="Authentication required")
db_path = await _ensure_db(request)
async with aiosqlite.connect(str(db_path)) as db:
db.row_factory = aiosqlite.Row
cursor = await db.execute("SELECT * FROM users WHERE id = ?", (user_id,))
row = await cursor.fetchone()
if row is None:
raise HTTPException(status_code=404, detail="User not found")
return _user_row_to_response(row)

0
tests/integration/auth/__init__.py Normal file
View File

275
tests/integration/auth/test_admin_routes.py Normal file
View File

@ -0,0 +1,275 @@
"""Integration tests for the admin session-management routes (U4).
Per the plan (U4 admin test scenarios):
- ``GET /admin/users/{id}/sessions`` as admin → returns all sessions (active + revoked)
- ``GET /admin/users/{id}/sessions`` as non-admin → 403
- ``DELETE /admin/users/{id}/sessions/{sid}`` as admin → that session revoked
- ``DELETE /admin/users/{id}/sessions/{sid}`` as non-admin → 403
"""
from __future__ import annotations
import uuid
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
import aiosqlite
import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient
from agentkit.server.auth.denylist import InMemoryRecentlyRevoked
from agentkit.server.auth.middleware import AuthMiddleware
from agentkit.server.auth.models import init_auth_db
from agentkit.server.auth.password import hash_password
from agentkit.server.auth.session_service import SessionService, set_session_service
from agentkit.server.routes import auth as auth_routes
# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------
@pytest.fixture
def jwt_secret() -> str:
return "admin-integration-test-jwt-secret-32bytes!"
@pytest.fixture
async def tmp_auth_db(tmp_path: Path) -> Path:
db_path = tmp_path / "admin_auth_integration.db"
await init_auth_db(db_path)
return db_path
async def _insert_user(
db_path: Path,
*,
username: str,
password: str,
role: str = "member",
) -> dict[str, Any]:
user_id = str(uuid.uuid4())
email = f"{username}@example.com"
password_hash = hash_password(password)
now_iso = datetime.now(timezone.utc).isoformat()
async with aiosqlite.connect(str(db_path)) as db:
await db.execute(
"INSERT INTO users "
"(id, username, email, password_hash, role, is_active, "
" is_terminal_authorized, is_server_terminal_authorized, "
" created_at, updated_at) "
"VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
(user_id, username, email, password_hash, role, 1, 0, 0, now_iso, now_iso),
)
await db.commit()
return {
"id": user_id,
"username": username,
"email": email,
"password": password,
"role": role,
}
@pytest.fixture
async def users(tmp_auth_db: Path) -> dict[str, dict[str, Any]]:
"""Create a member and an admin in the same DB."""
member = await _insert_user(tmp_auth_db, username="alice", password="alice-pw-12345")
admin = await _insert_user(
tmp_auth_db,
username="admin",
password="admin-pw-12345",
role="admin",
)
return {"member": member, "admin": admin}
@pytest.fixture
def auth_app(jwt_secret: str, users: dict[str, dict[str, Any]], tmp_auth_db: Path) -> FastAPI:
app = FastAPI()
app.state.jwt_secret = jwt_secret
app.state.auth_db_path = str(tmp_auth_db)
app.add_middleware(AuthMiddleware, jwt_secret=jwt_secret)
app.include_router(auth_routes.router, prefix="/api/v1")
app.include_router(auth_routes.admin_router, prefix="/api/v1")
return app
@pytest.fixture
def auth_client(auth_app: FastAPI) -> TestClient:
test_svc = SessionService(
db_path=auth_app.state.auth_db_path,
denylist=InMemoryRecentlyRevoked(),
)
set_session_service(test_svc)
try:
yield TestClient(auth_app)
finally:
set_session_service(None)
def _login(client: TestClient, username: str, password: str) -> dict[str, Any]:
resp = client.post(
"/api/v1/auth/login",
json={"username": username, "password": password},
)
assert resp.status_code == 200, resp.text
return resp.json()
# ---------------------------------------------------------------------------
# Admin: list user sessions
# ---------------------------------------------------------------------------
class TestAdminListUserSessions:
"""GET /admin/users/{user_id}/sessions."""
def test_admin_can_list_user_sessions(
self,
auth_client: TestClient,
users: dict[str, dict[str, Any]],
):
# Member logs in (creates a session).
_login(auth_client, users["member"]["username"], users["member"]["password"])
# Admin logs in.
admin_body = _login(
auth_client,
users["admin"]["username"],
users["admin"]["password"],
)
member_id = users["member"]["id"]
resp = auth_client.get(
f"/api/v1/admin/users/{member_id}/sessions",
headers={"Authorization": f"Bearer {admin_body['access_token']}"},
)
assert resp.status_code == 200, resp.text
sessions = resp.json()
assert len(sessions) >= 1
# All sessions belong to the member.
assert all(s["user_id"] == member_id for s in sessions)
def test_non_admin_cannot_list_user_sessions(
self,
auth_client: TestClient,
users: dict[str, dict[str, Any]],
):
# Member logs in.
member_body = _login(
auth_client,
users["member"]["username"],
users["member"]["password"],
)
# Member tries to access admin endpoint.
resp = auth_client.get(
f"/api/v1/admin/users/{users['member']['id']}/sessions",
headers={"Authorization": f"Bearer {member_body['access_token']}"},
)
assert resp.status_code in (403, 401)
# ---------------------------------------------------------------------------
# Admin: revoke user session
# ---------------------------------------------------------------------------
class TestAdminRevokeUserSession:
"""DELETE /admin/users/{user_id}/sessions/{session_id}."""
def test_admin_can_revoke_user_session(
self,
auth_client: TestClient,
users: dict[str, dict[str, Any]],
):
# Member logs in (creates a session).
member_body = _login(
auth_client,
users["member"]["username"],
users["member"]["password"],
)
# Get the member's session id via whoami.
whoami_resp = auth_client.get(
"/api/v1/auth/whoami",
headers={"Authorization": f"Bearer {member_body['access_token']}"},
)
assert whoami_resp.status_code == 200
member_sid = whoami_resp.json()["session_id"]
assert member_sid is not None
# Admin logs in and revokes the member's session.
admin_body = _login(
auth_client,
users["admin"]["username"],
users["admin"]["password"],
)
resp = auth_client.delete(
f"/api/v1/admin/users/{users['member']['id']}/sessions/{member_sid}",
headers={"Authorization": f"Bearer {admin_body['access_token']}"},
)
assert resp.status_code == 200, resp.text
assert resp.json()["revoked"] is True
# Member's session should now be invalid.
resp_member = auth_client.get(
"/api/v1/auth/whoami",
headers={"Authorization": f"Bearer {member_body['access_token']}"},
)
assert resp_member.status_code == 401
def test_non_admin_cannot_revoke_user_session(
self,
auth_client: TestClient,
users: dict[str, dict[str, Any]],
):
member_body = _login(
auth_client,
users["member"]["username"],
users["member"]["password"],
)
whoami_resp = auth_client.get(
"/api/v1/auth/whoami",
headers={"Authorization": f"Bearer {member_body['access_token']}"},
)
member_sid = whoami_resp.json()["session_id"]
# Member tries to use the admin endpoint.
resp = auth_client.delete(
f"/api/v1/admin/users/{users['member']['id']}/sessions/{member_sid}",
headers={"Authorization": f"Bearer {member_body['access_token']}"},
)
assert resp.status_code in (403, 401)
# ---------------------------------------------------------------------------
# Admin: list all sessions
# ---------------------------------------------------------------------------
class TestAdminListAllSessions:
"""GET /admin/sessions."""
def test_admin_can_list_all_sessions(
self,
auth_client: TestClient,
users: dict[str, dict[str, Any]],
):
# Both users log in.
_login(auth_client, users["member"]["username"], users["member"]["password"])
admin_body = _login(
auth_client,
users["admin"]["username"],
users["admin"]["password"],
)
resp = auth_client.get(
"/api/v1/admin/sessions",
headers={"Authorization": f"Bearer {admin_body['access_token']}"},
)
assert resp.status_code == 200, resp.text
sessions = resp.json()
# At least 2 sessions (member + admin).
assert len(sessions) >= 2

431
tests/integration/auth/test_auth_routes.py Normal file
View File

@ -0,0 +1,431 @@
"""Integration tests for the auth REST routes (U4).
These tests exercise the full HTTP stack (TestClient → AuthMiddleware →
route → SessionService → SQLite) end-to-end. They complement the unit
tests in ``tests/unit/test_auth.py`` by covering the new endpoints
introduced by the centralized-auth feature: ``/auth/whoami`` (with
refresh-token cold-start), ``/auth/sessions``, ``/auth/logout-others``,
``/auth/change-password``.
Per the plan (U4 test scenarios), the suite covers:
- Happy paths (login, refresh, whoami, sessions, logout-others, change-password)
- Error paths (wrong password, unknown user, expired token, missing auth)
- Integration (multi-device sessions, session cap eviction, password
change invalidating other devices)
"""
from __future__ import annotations
import uuid
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
import aiosqlite
import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient
from agentkit.server.auth.denylist import InMemoryRecentlyRevoked
from agentkit.server.auth.middleware import AuthMiddleware
from agentkit.server.auth.models import init_auth_db
from agentkit.server.auth.password import hash_password
from agentkit.server.auth.session_service import SessionService, set_session_service
from agentkit.server.routes import auth as auth_routes
# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------
@pytest.fixture
def jwt_secret() -> str:
return "integration-test-jwt-secret-do-not-use-in-prod-32bytes"
@pytest.fixture
async def tmp_auth_db(tmp_path: Path) -> Path:
db_path = tmp_path / "auth_integration.db"
await init_auth_db(db_path)
return db_path
async def _insert_user(
db_path: Path,
*,
username: str = "alice",
password: str = "correct-horse-battery-staple",
role: str = "member",
) -> dict[str, Any]:
"""Insert a user row and return the user fields + plaintext password."""
user_id = str(uuid.uuid4())
email = f"{username}@example.com"
password_hash = hash_password(password)
now_iso = datetime.now(timezone.utc).isoformat()
async with aiosqlite.connect(str(db_path)) as db:
await db.execute(
"INSERT INTO users "
"(id, username, email, password_hash, role, is_active, "
" is_terminal_authorized, is_server_terminal_authorized, "
" created_at, updated_at) "
"VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
(user_id, username, email, password_hash, role, 1, 0, 0, now_iso, now_iso),
)
await db.commit()
return {
"id": user_id,
"username": username,
"email": email,
"password": password,
"role": role,
"db_path": db_path,
}
@pytest.fixture
async def auth_db_with_user(tmp_auth_db: Path) -> dict[str, Any]:
return await _insert_user(tmp_auth_db)
@pytest.fixture
async def auth_db_with_admin(tmp_auth_db: Path) -> dict[str, Any]:
"""A second DB+user with admin role (for admin-route tests)."""
return await _insert_user(
tmp_auth_db,
username="admin",
password="admin-horse-battery-staple",
role="admin",
)
@pytest.fixture
def auth_app(jwt_secret: str, auth_db_with_user: dict[str, Any]) -> FastAPI:
app = FastAPI()
app.state.jwt_secret = jwt_secret
app.state.auth_db_path = str(auth_db_with_user["db_path"])
app.add_middleware(AuthMiddleware, jwt_secret=jwt_secret)
app.include_router(auth_routes.router, prefix="/api/v1")
app.include_router(auth_routes.admin_router, prefix="/api/v1")
return app
@pytest.fixture
def auth_client(auth_app: FastAPI) -> TestClient:
"""TestClient with SessionService bound to the test DB."""
test_svc = SessionService(
db_path=auth_app.state.auth_db_path,
denylist=InMemoryRecentlyRevoked(),
)
set_session_service(test_svc)
try:
yield TestClient(auth_app)
finally:
set_session_service(None)
def _login(client: TestClient, username: str, password: str) -> dict[str, Any]:
"""Helper: log in and return the TokenResponse body."""
resp = client.post(
"/api/v1/auth/login",
json={"username": username, "password": password},
)
assert resp.status_code == 200, resp.text
return resp.json()
# ---------------------------------------------------------------------------
# Happy paths
# ---------------------------------------------------------------------------
class TestWhoamiColdStart:
"""GET /auth/whoami with access OR refresh token (cold-start)."""
def test_whoami_with_access_token_returns_user(
self,
auth_client: TestClient,
auth_db_with_user: dict[str, Any],
):
body = _login(auth_client, auth_db_with_user["username"], auth_db_with_user["password"])
resp = auth_client.get(
"/api/v1/auth/whoami",
headers={"Authorization": f"Bearer {body['access_token']}"},
)
assert resp.status_code == 200, resp.text
data = resp.json()
assert data["user"]["username"] == auth_db_with_user["username"]
# Access-token call does NOT issue a new access token.
assert data["access_token"] is None
assert data["session_id"] is not None
assert data["session"] is not None
def test_whoami_with_refresh_token_issues_new_access_token(
self,
auth_client: TestClient,
auth_db_with_user: dict[str, Any],
):
"""Cold-start: refresh token → fresh access token returned."""
body = _login(auth_client, auth_db_with_user["username"], auth_db_with_user["password"])
resp = auth_client.get(
"/api/v1/auth/whoami",
headers={"Authorization": f"Bearer {body['refresh_token']}"},
)
assert resp.status_code == 200, resp.text
data = resp.json()
assert data["user"]["username"] == auth_db_with_user["username"]
# Cold-start issues a fresh access token.
assert data["access_token"] is not None
assert data["access_token"] != body["access_token"]
assert data["session_id"] is not None
def test_whoami_missing_bearer_returns_401(self, auth_client: TestClient):
resp = auth_client.get("/api/v1/auth/whoami")
assert resp.status_code == 401
def test_whoami_invalid_token_returns_401(self, auth_client: TestClient):
resp = auth_client.get(
"/api/v1/auth/whoami",
headers={"Authorization": "Bearer not.a.valid.jwt"},
)
assert resp.status_code == 401
class TestSessionsManagement:
"""GET /auth/sessions, DELETE /auth/sessions/{id}."""
def test_list_sessions_returns_current_user_sessions(
self,
auth_client: TestClient,
auth_db_with_user: dict[str, Any],
):
body = _login(auth_client, auth_db_with_user["username"], auth_db_with_user["password"])
resp = auth_client.get(
"/api/v1/auth/sessions",
headers={"Authorization": f"Bearer {body['access_token']}"},
)
assert resp.status_code == 200, resp.text
sessions = resp.json()
assert len(sessions) >= 1
# The current session should be marked is_current=True.
assert any(s["is_current"] for s in sessions)
def test_revoke_own_session(
self,
auth_client: TestClient,
auth_db_with_user: dict[str, Any],
):
body = _login(auth_client, auth_db_with_user["username"], auth_db_with_user["password"])
# List sessions to get the session id.
sessions = auth_client.get(
"/api/v1/auth/sessions",
headers={"Authorization": f"Bearer {body['access_token']}"},
).json()
sid = sessions[0]["id"]
resp = auth_client.delete(
f"/api/v1/auth/sessions/{sid}",
headers={"Authorization": f"Bearer {body['access_token']}"},
)
assert resp.status_code == 200, resp.text
assert resp.json()["revoked"] is True
def test_revoke_other_user_session_returns_404(
self,
auth_client: TestClient,
auth_db_with_user: dict[str, Any],
tmp_auth_db: Path,
):
"""A user cannot revoke another user's session (404, not 403, to avoid leaking)."""
# Create a second user and log in as them.
other = _login_sync_create_user(auth_client, tmp_auth_db, username="bob")
# Alice tries to revoke Bob's session.
body = _login(auth_client, auth_db_with_user["username"], auth_db_with_user["password"])
resp = auth_client.delete(
f"/api/v1/auth/sessions/{other['session_id']}",
headers={"Authorization": f"Bearer {body['access_token']}"},
)
assert resp.status_code == 404
def _login_sync_create_user(
client: TestClient,
db_path: Path,
*,
username: str,
) -> dict[str, Any]:
"""Insert a user directly into the DB and log in via the API.
Returns the login response + the session_id extracted from the JWT.
"""
import asyncio
from agentkit.server.auth.jwt_utils import verify_token
user = asyncio.run(_insert_user(db_path, username=username))
body = _login(client, username, user["password"])
payload = verify_token(body["access_token"], client.app.state.jwt_secret)
return {**body, "session_id": payload.get("sid"), "user_id": user["id"]}
class TestLogoutOthers:
"""POST /auth/logout-others."""
def test_logout_others_revokes_all_other_sessions(
self,
auth_client: TestClient,
auth_db_with_user: dict[str, Any],
tmp_auth_db: Path,
):
"""Login from 2 devices, logout-others from device A → B's session revoked."""
# Device A
body_a = _login(auth_client, auth_db_with_user["username"], auth_db_with_user["password"])
# Device B (same user, different fingerprint header)
body_b = _login(auth_client, auth_db_with_user["username"], auth_db_with_user["password"])
# A calls logout-others.
resp = auth_client.post(
"/api/v1/auth/logout-others",
headers={"Authorization": f"Bearer {body_a['access_token']}"},
)
assert resp.status_code == 200, resp.text
assert resp.json()["revoked_count"] >= 1
# B's session should now be revoked → whoami with B's refresh token → 401.
resp_b = auth_client.get(
"/api/v1/auth/whoami",
headers={"Authorization": f"Bearer {body_b['refresh_token']}"},
)
assert resp_b.status_code == 401
# A's session should still be valid.
resp_a = auth_client.get(
"/api/v1/auth/whoami",
headers={"Authorization": f"Bearer {body_a['access_token']}"},
)
assert resp_a.status_code == 200
class TestChangePassword:
"""POST /auth/change-password."""
def test_change_password_revokes_other_sessions(
self,
auth_client: TestClient,
auth_db_with_user: dict[str, Any],
):
"""Change password → other sessions revoked, current stays alive."""
# Device A
body_a = _login(auth_client, auth_db_with_user["username"], auth_db_with_user["password"])
# Device B
body_b = _login(auth_client, auth_db_with_user["username"], auth_db_with_user["password"])
# A changes the password.
resp = auth_client.post(
"/api/v1/auth/change-password",
headers={"Authorization": f"Bearer {body_a['access_token']}"},
json={
"old_password": auth_db_with_user["password"],
"new_password": "new-strong-password-123",
},
)
assert resp.status_code == 200, resp.text
# B's session should be revoked.
resp_b = auth_client.get(
"/api/v1/auth/whoami",
headers={"Authorization": f"Bearer {body_b['refresh_token']}"},
)
assert resp_b.status_code == 401
def test_change_password_wrong_old_returns_400(
self,
auth_client: TestClient,
auth_db_with_user: dict[str, Any],
):
body = _login(auth_client, auth_db_with_user["username"], auth_db_with_user["password"])
resp = auth_client.post(
"/api/v1/auth/change-password",
headers={"Authorization": f"Bearer {body['access_token']}"},
json={
"old_password": "totally-wrong",
"new_password": "new-strong-password-123",
},
)
assert resp.status_code == 400
def test_change_password_weak_new_returns_400(
self,
auth_client: TestClient,
auth_db_with_user: dict[str, Any],
):
body = _login(auth_client, auth_db_with_user["username"], auth_db_with_user["password"])
resp = auth_client.post(
"/api/v1/auth/change-password",
headers={"Authorization": f"Bearer {body['access_token']}"},
json={
"old_password": auth_db_with_user["password"],
"new_password": "short",
},
)
assert resp.status_code == 400
# ---------------------------------------------------------------------------
# Integration: multi-device + session cap
# ---------------------------------------------------------------------------
class TestMultiDeviceIntegration:
"""Login from multiple devices → independent sessions."""
def test_multiple_logins_create_independent_sessions(
self,
auth_client: TestClient,
auth_db_with_user: dict[str, Any],
):
body_a = _login(auth_client, auth_db_with_user["username"], auth_db_with_user["password"])
body_b = _login(auth_client, auth_db_with_user["username"], auth_db_with_user["password"])
# Both tokens should work independently.
resp_a = auth_client.get(
"/api/v1/auth/whoami",
headers={"Authorization": f"Bearer {body_a['access_token']}"},
)
resp_b = auth_client.get(
"/api/v1/auth/whoami",
headers={"Authorization": f"Bearer {body_b['access_token']}"},
)
assert resp_a.status_code == 200
assert resp_b.status_code == 200
# Different session ids.
assert resp_a.json()["session_id"] != resp_b.json()["session_id"]
def test_session_cap_evicts_oldest(
self,
auth_client: TestClient,
auth_db_with_user: dict[str, Any],
):
"""The default session cap is 10; the 11th login evicts the oldest."""
tokens = []
for _ in range(11):
body = _login(
auth_client,
auth_db_with_user["username"],
auth_db_with_user["password"],
)
tokens.append(body)
# The first session should have been evicted (whoami → 401).
resp_first = auth_client.get(
"/api/v1/auth/whoami",
headers={"Authorization": f"Bearer {tokens[0]['refresh_token']}"},
)
assert resp_first.status_code == 401
# The last session should still be valid.
resp_last = auth_client.get(
"/api/v1/auth/whoami",
headers={"Authorization": f"Bearer {tokens[-1]['access_token']}"},
)
assert resp_last.status_code == 200