NgTech-LTD/gns3-server
2
0
Fork 0
mirror of https://github.com/GNS3/gns3-server.git synced 2026-08-27 12:30:13 +03:00
Code Issues Packages Projects Releases Wiki Activity

Generate fresh JWT on API key auth instead of returning raw key

Browse Source 
When MCP client authenticates with API key, _resolve_token now
returns a newly generated JWT. This JWT is stored in the ContextVar
and used by all subsequent tool handler API calls - zero extra bcrypt.
This commit is contained in:
YueGuobin 2026-06-16 00:09:36 +08:00
parent f522c947bc
commit d6b337bc3a
No known key found for this signature in database
1 changed files with 17 additions and 13 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

30
gns3server/api/routes/mcp/__init__.py
View File

@ -213,23 +213,27 @@ async def _resolve_token(token: str) -> str | None:
except Exception:
pass
# Try API key
# Try API key — format: gns3_<api_key_id>_<random_secret> → O(1) lookup
if token.startswith("gns3_") and _app is not None:
db_engine = getattr(_app.state, "_db_engine", None)
if db_engine is not None:
try:
async with AsyncSession(db_engine, expire_on_commit=False) as db_session:
repo = ApiKeysRepository(db_session)
query = select(models.ApiKey).where(models.ApiKey.revoked == False)
result = await db_session.execute(query)
for db_key in result.scalars().all():
if bcrypt.checkpw(token.encode(), db_key.key_hash.encode()):
await repo.update_last_used(db_key.api_key_id)
user_repo = UsersRepository(db_session)
user = await user_repo.get_user(db_key.user_id)
if user:
_jwt_username_var.set(user.username)
return token
parts = token.split("_", 2)
if len(parts) == 3:
key_id = UUID(parts[1])
secret = parts[2]
async with AsyncSession(db_engine, expire_on_commit=False) as db_session:
repo = ApiKeysRepository(db_session)
db_key = await repo.get_api_key(key_id)
if db_key and not db_key.revoked:
if await asyncio.to_thread(bcrypt.checkpw, secret.encode(), db_key.key_hash.encode()):
await repo.update_last_used(db_key.api_key_id)
user_repo = UsersRepository(db_session)
user = await user_repo.get_user(db_key.user_id)
if user:
_jwt_username_var.set(user.username)
fresh_token = auth_service.create_access_token(user.username)
return fresh_token
except Exception:
pass