feat: add limited visitor tokens
This commit is contained in:
@@ -0,0 +1,14 @@
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from app.main import create_app
|
||||
|
||||
|
||||
def test_visitor_token_endpoint_returns_short_lived_bearer_token() -> None:
|
||||
with TestClient(create_app()) as client:
|
||||
response = client.post("/api/v1/visitor-tokens")
|
||||
|
||||
assert response.status_code == 201
|
||||
body = response.json()
|
||||
assert isinstance(body["access_token"], str)
|
||||
assert body["token_type"] == "Bearer"
|
||||
assert body["expires_in"] == 900
|
||||
@@ -6,7 +6,7 @@ import pytest
|
||||
|
||||
from app.core.config import Settings
|
||||
from app.core.errors import UnauthorizedAgentError
|
||||
from app.core.security import JwtAuthenticator
|
||||
from app.core.security import JwtAuthenticator, VisitorTokenIssuer
|
||||
|
||||
|
||||
def _settings() -> Settings:
|
||||
@@ -39,6 +39,34 @@ def test_authenticate_valid_token() -> None:
|
||||
assert context.trace_id
|
||||
|
||||
|
||||
def test_authenticate_visitor_token_returns_limited_anonymous_context() -> None:
|
||||
now = datetime.now(UTC)
|
||||
private_key = Path("config/jwt/jwt-private.pem").read_text(encoding="utf-8")
|
||||
token = jwt.encode(
|
||||
{"sub": "2", "iss": "jr-auth", "aud": "jr-agent-platform", "iat": now,
|
||||
"nbf": now, "exp": now + timedelta(minutes=5), "jti": "visitor-jti-1",
|
||||
"visitor": True},
|
||||
private_key,
|
||||
algorithm="RS256",
|
||||
)
|
||||
|
||||
context = JwtAuthenticator(_settings()).authenticate(token)
|
||||
|
||||
assert context.roles == ("visitor",)
|
||||
assert context.permissions == ("agent:run",)
|
||||
assert context.customer_ids == ()
|
||||
|
||||
|
||||
def test_visitor_token_issuer_creates_short_lived_limited_token() -> None:
|
||||
token, expires_at = VisitorTokenIssuer(_settings()).issue()
|
||||
|
||||
context = JwtAuthenticator(_settings()).authenticate(token)
|
||||
|
||||
assert context.roles == ("visitor",)
|
||||
assert context.permissions == ("agent:run",)
|
||||
assert expires_at > datetime.now(UTC)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("subject", ["abc", "", "0", "-1", "1.5", "12", "1" * 21,
|
||||
"18446744073709551616"])
|
||||
def test_invalid_numeric_subject_is_unauthorized(subject):
|
||||
@@ -62,6 +90,37 @@ def test_signed_invalid_subject_returns_401_before_identity_query(monkeypatch):
|
||||
resolve.assert_not_awaited()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_visitor_token_skips_identity_database_resolution(monkeypatch):
|
||||
from unittest.mock import AsyncMock
|
||||
|
||||
from fastapi.security import HTTPAuthorizationCredentials
|
||||
from starlette.requests import Request
|
||||
|
||||
from app.api.dependencies.auth import _authenticator, build_request_context
|
||||
|
||||
now = datetime.now(UTC)
|
||||
private_key = Path("config/jwt/jwt-private.pem").read_text(encoding="utf-8")
|
||||
token = jwt.encode(
|
||||
{"sub": "2", "iss": "jr-local", "aud": "jr-agent-platform", "iat": now,
|
||||
"nbf": now, "exp": now + timedelta(minutes=5), "jti": "visitor-jti-2",
|
||||
"visitor": True},
|
||||
private_key,
|
||||
algorithm="RS256",
|
||||
)
|
||||
resolve = AsyncMock()
|
||||
monkeypatch.setattr("app.service.identity_service.IdentityService.resolve", resolve)
|
||||
_authenticator.cache_clear()
|
||||
request = Request({"type": "http", "method": "GET", "path": "/"})
|
||||
|
||||
context = await build_request_context(
|
||||
request, HTTPAuthorizationCredentials(scheme="Bearer", credentials=token)
|
||||
)
|
||||
|
||||
assert context.roles == ("visitor",)
|
||||
resolve.assert_not_awaited()
|
||||
|
||||
|
||||
def test_authenticate_rejects_expired_token() -> None:
|
||||
with pytest.raises(UnauthorizedAgentError):
|
||||
now = datetime.now(UTC)
|
||||
|
||||
Reference in New Issue
Block a user