feat(session): add SessionTokenManager with token creation
Add SessionTokenManager class that creates short-lived session tokens for HTTP proxy access. Each token includes agent identity, document scope, permissions, and expiration time.
This commit is contained in:
46
src/grist_mcp/session.py
Normal file
46
src/grist_mcp/session.py
Normal file
@@ -0,0 +1,46 @@
|
||||
"""Session token management for HTTP proxy access."""
|
||||
|
||||
import secrets
|
||||
from dataclasses import dataclass
|
||||
from datetime import datetime, timedelta, timezone
|
||||
|
||||
|
||||
@dataclass
|
||||
class SessionToken:
|
||||
"""A short-lived session token for proxy access."""
|
||||
token: str
|
||||
document: str
|
||||
permissions: list[str]
|
||||
agent_name: str
|
||||
created_at: datetime
|
||||
expires_at: datetime
|
||||
|
||||
|
||||
class SessionTokenManager:
|
||||
"""Manages creation and validation of session tokens."""
|
||||
|
||||
def __init__(self):
|
||||
self._tokens: dict[str, SessionToken] = {}
|
||||
|
||||
def create_token(
|
||||
self,
|
||||
agent_name: str,
|
||||
document: str,
|
||||
permissions: list[str],
|
||||
ttl_seconds: int,
|
||||
) -> SessionToken:
|
||||
"""Create a new session token."""
|
||||
now = datetime.now(timezone.utc)
|
||||
token_str = f"sess_{secrets.token_urlsafe(32)}"
|
||||
|
||||
session = SessionToken(
|
||||
token=token_str,
|
||||
document=document,
|
||||
permissions=permissions,
|
||||
agent_name=agent_name,
|
||||
created_at=now,
|
||||
expires_at=now + timedelta(seconds=ttl_seconds),
|
||||
)
|
||||
|
||||
self._tokens[token_str] = session
|
||||
return session
|
||||
23
tests/unit/test_session.py
Normal file
23
tests/unit/test_session.py
Normal file
@@ -0,0 +1,23 @@
|
||||
import pytest
|
||||
from datetime import datetime, timedelta, timezone
|
||||
|
||||
from grist_mcp.session import SessionTokenManager, SessionToken
|
||||
|
||||
|
||||
def test_create_token_returns_valid_session_token():
|
||||
manager = SessionTokenManager()
|
||||
|
||||
token = manager.create_token(
|
||||
agent_name="test-agent",
|
||||
document="sales",
|
||||
permissions=["read", "write"],
|
||||
ttl_seconds=300,
|
||||
)
|
||||
|
||||
assert token.token.startswith("sess_")
|
||||
assert len(token.token) > 20
|
||||
assert token.document == "sales"
|
||||
assert token.permissions == ["read", "write"]
|
||||
assert token.agent_name == "test-agent"
|
||||
assert token.expires_at > datetime.now(timezone.utc)
|
||||
assert token.expires_at < datetime.now(timezone.utc) + timedelta(seconds=310)
|
||||
Reference in New Issue
Block a user