583 lines
33 KiB
Python
583 lines
33 KiB
Python
"""Exercise extracted skills, real HTTP boundaries and MCP stdio without a paid LLM."""
|
|
import asyncio
|
|
import copy
|
|
import hashlib
|
|
import importlib.util
|
|
import io
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import subprocess
|
|
import sys
|
|
import threading
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
import urllib.request
|
|
import zipfile
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
from neo4j.exceptions import ServiceUnavailable
|
|
|
|
import app.main as main
|
|
import app.routes as routes
|
|
from scripts.build_claude_packages import build_package
|
|
from packages.claude import client
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
KEY = 'a' * 64
|
|
|
|
|
|
@pytest.fixture
|
|
def backend(monkeypatch):
|
|
monkeypatch.setenv('ZP_MCP_AUTH_MODE', 'bearer')
|
|
monkeypatch.setattr(main, 'embed_query', lambda _: None)
|
|
evidence = {'status': 'ok', 'reason': None, 'complete': False, 'truncated': True,
|
|
'truncation_reasons': ['result_limit'], 'index_version': {'version': 'fixture'},
|
|
'sources': [{'source_id': 'S1', 'title': 'Eva Nová', 'document_path': 'pages/eva/README.md',
|
|
'source_url': 'https://example.test/eva'}],
|
|
'context': 'Eva Nová: Diplomová práca 2024.',
|
|
'facts': [{'person': 'Eva Nová', 'year': 2024}], 'paths': []}
|
|
checked = {k: copy.deepcopy(v) for k, v in evidence.items() if k not in {'sources', 'context', 'facts', 'paths'}}
|
|
checked.update(accepted_for_delivery=True, answer='Eva Nová — rok práce: 2024.\nVýpis je neúplný.',
|
|
claims=[{'person': 'Eva Nová', 'year': 2024}], source_urls=['https://example.test/eva'],
|
|
attempts=[{'raw_response': 'PRIVATE MODEL DIAGNOSTICS'}])
|
|
monkeypatch.setattr(routes, 'deliver_answer', lambda *args, **kw: copy.deepcopy(checked))
|
|
monkeypatch.setattr(routes, 'search_graph', lambda *args, **kw: copy.deepcopy(evidence))
|
|
monkeypatch.setattr(routes, 'build_rag_context', lambda *args, **kw: {
|
|
**{k: v for k, v in evidence.items() if k in routes.RagResponse.model_fields},
|
|
'query': args[1], 'engine': 'hybrid_fts5_embeddings', 'source_count': 1,
|
|
'strategies': [], 'instructions': [], 'answer_format': {}})
|
|
monkeypatch.setattr(routes, 'compact_rag_payload', lambda value, **kw: value)
|
|
state = {'raw': None, 'code': 200, 'requests': [], 'checked': checked, 'evidence': evidence}
|
|
with TestClient(main.app) as api:
|
|
class Handler(BaseHTTPRequestHandler):
|
|
def log_message(self, *args):
|
|
pass
|
|
|
|
def do_GET(self):
|
|
value = {'ready': True, 'index_version': {'version': 'fixture'}, 'tools': {
|
|
engine: {'ready': True, 'index_version': {'version': 'fixture'}, 'reason': None}
|
|
for engine in ['rag', 'graphrag']}}
|
|
self.send_response(200)
|
|
self.end_headers()
|
|
self.wfile.write(json.dumps(value).encode())
|
|
|
|
def do_POST(self):
|
|
body = self.rfile.read(int(self.headers.get('Content-Length', '0')))
|
|
state['requests'].append((self.path, json.loads(body), self.headers.get('Authorization')))
|
|
if state['raw'] is None:
|
|
response = api.post(self.path, content=body, headers=dict(self.headers))
|
|
code, raw = response.status_code, response.content
|
|
else:
|
|
code, raw = state['code'], state['raw']
|
|
self.send_response(code)
|
|
if code == 302:
|
|
self.send_header('Location', state['url'] + '/credential-leak')
|
|
self.end_headers()
|
|
self.wfile.write(raw)
|
|
|
|
server = ThreadingHTTPServer(('127.0.0.1', 0), Handler)
|
|
thread = threading.Thread(target=server.serve_forever, daemon=True)
|
|
thread.start()
|
|
state['url'] = f'http://127.0.0.1:{server.server_port}'
|
|
monkeypatch.setenv('ZP_TEST_HTTP_PORT', str(server.server_port))
|
|
monkeypatch.setenv('ZP_AGENT_URL', state['url'])
|
|
monkeypatch.setenv('ZP_AGENT_API_KEY', KEY)
|
|
monkeypatch.delenv('ZP_AGENT_CONFIG', raising=False)
|
|
try:
|
|
yield state
|
|
finally:
|
|
server.shutdown()
|
|
server.server_close()
|
|
thread.join()
|
|
|
|
|
|
@pytest.mark.parametrize('engine', ['rag', 'graphrag'])
|
|
def test_answer_is_exact_checked_text_without_diagnostics(backend, engine):
|
|
result = client.query_backend('Práca Evy Novej?', engine)
|
|
assert result['accepted_for_delivery'] is True
|
|
assert result['answer'] == backend['checked']['answer']
|
|
assert result['complete'] is False and result['truncated'] is True
|
|
assert result['index_version'] == {'version': 'fixture'}
|
|
assert 'attempts' not in result and 'PRIVATE' not in json.dumps(result)
|
|
path, body, authorization = backend['requests'][0]
|
|
assert path == '/answer' and body['engine'] == engine and body['include_diagnostics'] is False
|
|
assert authorization == 'Bearer ' + KEY
|
|
|
|
|
|
@pytest.mark.parametrize('engine', ['rag', 'graphrag'])
|
|
def test_retrieval_explicitly_requires_final_verification(backend, engine):
|
|
result = client.query_backend('Práca Evy Novej?', engine, mode='retrieve')
|
|
assert result['status'] == 'ok'
|
|
assert result['requires_final_answer_verification'] is True
|
|
assert result['sources'] and 'answer' not in result
|
|
assert backend['requests'][0][1]['published_only'] is True
|
|
|
|
|
|
@pytest.mark.parametrize('code,reason', [(401, 'unauthorized'), (403, 'forbidden'),
|
|
(503, 'backend_unavailable'), (302, 'redirect_not_allowed')])
|
|
def test_transport_errors_and_redirects_do_not_disclose_payload(backend, code, reason):
|
|
backend.update(raw=b'SECRET ERROR', code=code)
|
|
value = client.query_backend('Práca?', 'rag')
|
|
assert value['reason'] == reason and value['answer'] == ''
|
|
assert len(backend['requests']) == 1 and 'SECRET' not in json.dumps(value)
|
|
|
|
|
|
@pytest.mark.parametrize('change', [
|
|
{'accepted_for_delivery': False}, {'accepted_for_delivery': 'true'},
|
|
{'complete': True}, {'truncation_reasons': 'limit'}, {'status': 'unexpected'},
|
|
{'answer': ''}, {'source_urls': 'https://example.test'}, {'claims': None}])
|
|
def test_rejected_or_malformed_answer_is_never_delivered(backend, change):
|
|
backend.update(raw=json.dumps({**backend['checked'], **change}).encode())
|
|
value = client.query_backend('Práca?', 'rag')
|
|
assert value['status'] == 'error' and not value['accepted_for_delivery']
|
|
assert value['answer'] == '' and 'Eva' not in json.dumps(value)
|
|
|
|
|
|
def test_invalid_json_and_oversized_response(backend):
|
|
for raw, reason in [(b'not json', 'invalid_backend_response'),
|
|
(b'x' * (client.MAX_RESPONSE_BYTES + 1), 'response_too_large')]:
|
|
backend['raw'] = raw
|
|
assert client.query_backend('Práca?', 'rag')['reason'] == reason
|
|
|
|
|
|
def test_actual_backend_denies_wrong_key_and_health_keeps_versions(backend):
|
|
invalid = client.Config(backend['url'], 'wrong-test-key')
|
|
assert client.query_backend('Práca?', 'rag', config=invalid)['reason'] == 'unauthorized'
|
|
health = client.backend_health()
|
|
assert health['ready'] and all(value['ready'] for value in health['tools'].values())
|
|
assert health['tools']['rag']['index_version'] == health['tools']['graphrag']['index_version']
|
|
|
|
|
|
@pytest.mark.parametrize('status', ['no_answer', 'ambiguous'])
|
|
def test_negative_and_ambiguous_answers_stay_distinct(backend, status):
|
|
backend['checked'].update(status=status, answer='Žiadne doložené údaje.' if status == 'no_answer' else 'Spresni meno.',
|
|
claims=[], source_urls=[])
|
|
result = client.query_backend('Práca?', 'rag')
|
|
assert result['status'] == status and result['accepted_for_delivery'] is True
|
|
|
|
|
|
@pytest.mark.parametrize('query', ['', ' ', 'x' * 501, None])
|
|
def test_invalid_query_does_not_reach_backend(backend, query):
|
|
assert client.query_backend(query, 'rag')['reason'] == 'invalid_query'
|
|
assert backend['requests'] == []
|
|
|
|
|
|
@pytest.mark.parametrize('url', ['http://remote.test', 'https://user:password@example.test',
|
|
'https://example.test/?key=secret', 'https://example.test/#x', 'file:///tmp/x'])
|
|
def test_unsafe_configuration_is_rejected_without_network(url):
|
|
with pytest.raises(client.ClientError):
|
|
client.Config(url, KEY).validate()
|
|
|
|
|
|
def test_config_file_permissions_and_environment_override(tmp_path, monkeypatch):
|
|
path = tmp_path / 'config.json'
|
|
path.write_text(json.dumps({'backend_url': 'https://example.test', 'api_key': KEY}))
|
|
path.chmod(0o644)
|
|
monkeypatch.setenv('ZP_AGENT_CONFIG', str(path))
|
|
monkeypatch.delenv('ZP_AGENT_URL', raising=False)
|
|
monkeypatch.delenv('ZP_AGENT_API_KEY', raising=False)
|
|
with pytest.raises(client.ClientError, match='0600'):
|
|
client.Config.load()
|
|
path.chmod(0o600)
|
|
assert client.Config.load().key == KEY
|
|
monkeypatch.setenv('ZP_AGENT_URL', 'http://localhost:8000')
|
|
assert client.Config.load().url == 'http://localhost:8000'
|
|
|
|
|
|
@pytest.mark.parametrize('engine', ['rag', 'graphrag'])
|
|
def test_zip_is_reproducible_and_runs_without_repository(tmp_path, backend, engine):
|
|
one = build_package(engine, tmp_path / 'one')
|
|
two = build_package(engine, tmp_path / 'two')
|
|
assert one['sha256'] == two['sha256']
|
|
archive_path = tmp_path / 'one' / one['archive']
|
|
folder = tmp_path / 'installed'
|
|
with zipfile.ZipFile(archive_path) as archive:
|
|
assert len(archive.namelist()) == 4
|
|
manifest = json.loads(archive.read(f'zpwiki-{engine}/manifest.json'))
|
|
for name, digest in manifest['files'].items():
|
|
assert hashlib.sha256(archive.read(f'zpwiki-{engine}/' + name)).hexdigest() == digest
|
|
archive.extractall(folder)
|
|
environment = {**os.environ, 'PYTHONPATH': '', 'PYTHONDONTWRITEBYTECODE': '1'}
|
|
command = [sys.executable, str(folder / f'zpwiki-{engine}/scripts/query.py'), 'Práca Evy Novej?']
|
|
result = subprocess.run(command, cwd=folder, env=environment, capture_output=True, text=True, timeout=15)
|
|
assert result.returncode == 0, result.stderr
|
|
assert json.loads(result.stdout)['answer'] == backend['checked']['answer']
|
|
|
|
|
|
def test_committed_archives_match_current_sources(tmp_path):
|
|
manifest = json.loads((ROOT / 'packages/claude/dist/packages.json').read_text())
|
|
for row in manifest['packages']:
|
|
engine = row['name'].removeprefix('zpwiki-')
|
|
rebuilt = build_package(engine, tmp_path)
|
|
actual = ROOT / 'packages/claude/dist' / row['archive']
|
|
assert hashlib.sha256(actual.read_bytes()).hexdigest() == rebuilt['sha256'] == row['sha256']
|
|
|
|
|
|
def test_mcp_stdio_initialization_discovery_call_and_rejection(backend):
|
|
from mcp import ClientSession, StdioServerParameters
|
|
from mcp.client.stdio import stdio_client
|
|
|
|
async def run():
|
|
params = StdioServerParameters(command=sys.executable,
|
|
args=[str(ROOT / 'packages/claude/mcp_server.py')],
|
|
env={**os.environ, 'PYTHONDONTWRITEBYTECODE': '1'})
|
|
async with stdio_client(params) as (read, write):
|
|
async with ClientSession(read, write) as session:
|
|
initialized = await session.initialize()
|
|
assert initialized.serverInfo.name == 'ZP Wiki'
|
|
listed = await session.list_tools()
|
|
assert {row.name for row in listed.tools} == {'zpwiki_rag', 'zpwiki_graphrag', 'zpwiki_health'}
|
|
for engine in ['rag', 'graphrag']:
|
|
result = await session.call_tool('zpwiki_' + engine, {'query': 'Práca Evy Novej?'})
|
|
assert not result.isError
|
|
assert result.structuredContent['answer'] == backend['checked']['answer']
|
|
rejected = await session.call_tool('zpwiki_rag', {'query': ''})
|
|
assert rejected.isError and rejected.structuredContent['answer'] == ''
|
|
asyncio.run(run())
|
|
|
|
|
|
@pytest.mark.parametrize('provider', ['anthropic', 'openai_compatible'])
|
|
def test_model_adapter_request_and_response_shapes(monkeypatch, provider):
|
|
from tools.answers.model_client import complete_with_configured_model
|
|
calls = []
|
|
monkeypatch.setenv('ANSWER_PROVIDER', provider)
|
|
monkeypatch.setenv('ANSWER_API_KEY', 'test-secret')
|
|
monkeypatch.setenv('ANSWER_MODEL', 'explicit-test-model')
|
|
monkeypatch.setenv('ANSWER_LLM_URL', 'https://model.example.test/api')
|
|
content = '{"status":"no_answer"}'
|
|
payload = {'stop_reason': 'end_turn', 'content': [{'type': 'text', 'text': content}]} if provider == 'anthropic' else {
|
|
'choices': [{'message': {'content': content}}]}
|
|
|
|
class Opener:
|
|
def open(self, request, **kw):
|
|
calls.append((request, json.loads(request.data)))
|
|
return io.BytesIO(json.dumps(payload).encode())
|
|
monkeypatch.setattr(urllib.request, 'build_opener', lambda *args: Opener())
|
|
value = complete_with_configured_model([{'role': 'system', 'content': 'Rules'},
|
|
{'role': 'user', 'content': 'Question'}])
|
|
assert value['choices'][0]['message']['content'] == content
|
|
request, body = calls[0]
|
|
assert body['model'] == 'explicit-test-model'
|
|
if provider == 'anthropic':
|
|
assert body['system'] == 'Rules' and body['messages'][0]['role'] == 'user'
|
|
assert request.get_header('X-api-key') == 'test-secret'
|
|
else:
|
|
assert body['messages'][0]['role'] == 'system'
|
|
assert request.get_header('Authorization') == 'Bearer test-secret'
|
|
|
|
|
|
def test_anthropic_truncated_output_is_rejected(monkeypatch):
|
|
from tools.answers.model_client import complete_with_configured_model
|
|
monkeypatch.setenv('ANSWER_PROVIDER', 'anthropic')
|
|
monkeypatch.setenv('ANSWER_API_KEY', KEY)
|
|
monkeypatch.setenv('ANSWER_MODEL', 'explicit-test-model')
|
|
monkeypatch.delenv('ANSWER_LLM_URL', raising=False)
|
|
class Opener:
|
|
def open(self, *args, **kwargs):
|
|
return io.BytesIO(b'{"stop_reason":"max_tokens","content":[]}')
|
|
monkeypatch.setattr(urllib.request, 'build_opener', lambda *args: Opener())
|
|
with pytest.raises(RuntimeError, match='Incomplete'):
|
|
complete_with_configured_model([{'role': 'user', 'content': 'Question'}])
|
|
|
|
|
|
def test_remote_mcp_authentication_and_full_protocol(backend, monkeypatch):
|
|
import httpx
|
|
from mcp import ClientSession
|
|
from mcp.client.streamable_http import streamable_http_client
|
|
from packages.claude.mcp_server import create_http_app
|
|
token = 'm' * 64
|
|
monkeypatch.setenv('ZP_MCP_ACCESS_TOKEN', token)
|
|
monkeypatch.setenv('ZP_MCP_PUBLIC_URL', 'https://mcp.example.test/mcp')
|
|
app = create_http_app()
|
|
|
|
async def run():
|
|
async with app.app.router.lifespan_context(app.app):
|
|
transport = httpx.ASGITransport(app=app)
|
|
async with httpx.AsyncClient(transport=transport, base_url='https://mcp.example.test') as http:
|
|
for headers in [{}, {'Authorization': 'Bearer wrong'}, {'Authorization': 'Bearer ' + KEY}]:
|
|
denied = await http.post('/mcp', json={}, headers=headers)
|
|
assert denied.status_code == 401 and denied.json() == {'error': 'unauthorized'}
|
|
async with httpx.AsyncClient(transport=transport, base_url='https://mcp.example.test',
|
|
headers={'Authorization': 'Bearer ' + token}) as http:
|
|
hostile = await http.post('/mcp', json={}, headers={'Host': 'untrusted.example.test',
|
|
'Accept': 'application/json, text/event-stream'})
|
|
assert hostile.status_code == 421
|
|
async with streamable_http_client('https://mcp.example.test/mcp', http_client=http) as (read, write, _):
|
|
async with ClientSession(read, write) as session:
|
|
await session.initialize()
|
|
assert len((await session.list_tools()).tools) == 3
|
|
for engine in ['rag', 'graphrag']:
|
|
result = await session.call_tool('zpwiki_' + engine, {'query': 'Práca Evy Novej?'})
|
|
assert not result.isError
|
|
assert result.structuredContent['answer'] == backend['checked']['answer']
|
|
health = await session.call_tool('zpwiki_health')
|
|
assert not health.isError and health.structuredContent['ready']
|
|
asyncio.run(run())
|
|
|
|
|
|
@pytest.mark.parametrize('token,url', [('', 'https://mcp.example.test/mcp'),
|
|
('short', 'https://mcp.example.test/mcp'), (KEY, 'https://mcp.example.test/mcp'),
|
|
('m' * 64, 'http://mcp.example.test/mcp'), ('m' * 64, 'https://mcp.example.test/other')])
|
|
def test_remote_mcp_refuses_insecure_startup(backend, monkeypatch, token, url):
|
|
from packages.claude.mcp_server import create_http_app
|
|
monkeypatch.setenv('ZP_MCP_ACCESS_TOKEN', token)
|
|
monkeypatch.setenv('ZP_MCP_PUBLIC_URL', url)
|
|
with pytest.raises(ValueError):
|
|
create_http_app()
|
|
|
|
|
|
def test_graph_serialization_errors_have_shared_contract(monkeypatch):
|
|
monkeypatch.setattr(routes, 'search_graph', lambda *args, **kwargs: {'invalid': object()})
|
|
with TestClient(main.app) as client:
|
|
response = client.post('/graphrag', json={'query': 'Práca?'}, headers={'X-API-Key': KEY})
|
|
assert response.status_code == 500 and response.json()['reason'] == 'internal_error'
|
|
|
|
|
|
def test_graph_api_uses_shared_error_contract_and_validates_query(monkeypatch):
|
|
def unavailable(*args, **kwargs):
|
|
raise ServiceUnavailable('PRIVATE_DATABASE_ERROR')
|
|
monkeypatch.setattr(routes, 'search_graph', unavailable)
|
|
with TestClient(main.app) as client:
|
|
response = client.post('/graphrag', json={'query': 'Práca?'}, headers={'X-API-Key': KEY})
|
|
assert response.status_code == 503 and response.json()['reason'] == 'retrieval_unavailable'
|
|
assert 'PRIVATE_DATABASE_ERROR' not in response.text
|
|
response = client.post('/graphrag', json={'query': ' ', 'limit': 0}, headers={'X-API-Key': KEY})
|
|
assert response.status_code == 422 and response.json()['status'] == 'error'
|
|
assert client.post('/graphrag', json={'query': 'Práca?'}).status_code == 401
|
|
|
|
|
|
@pytest.fixture
|
|
def oauth_app(backend, monkeypatch, tmp_path):
|
|
from packages.claude.mcp_server import create_http_app
|
|
monkeypatch.setenv('ZP_MCP_AUTH_MODE', 'oauth')
|
|
monkeypatch.setenv('ZP_MCP_ACCESS_TOKEN', 'm' * 64)
|
|
monkeypatch.setenv('ZP_MCP_LOGIN_PASSWORD', 'p' * 64)
|
|
monkeypatch.setenv('ZP_MCP_PUBLIC_URL', 'https://mcp.example.test/mcp')
|
|
monkeypatch.setenv('ZP_MCP_OAUTH_DB', str(tmp_path/'oauth'/'oauth.sqlite'))
|
|
return create_http_app()
|
|
|
|
|
|
async def oauth_start(http, *, state='state-123', resource='https://mcp.example.test/mcp', auth_method='none'):
|
|
import base64
|
|
from urllib.parse import urlsplit,parse_qs
|
|
verifier='v' * 64
|
|
challenge=base64.urlsafe_b64encode(hashlib.sha256(verifier.encode()).digest()).decode().rstrip('=')
|
|
registered=await http.post('/register',json={'redirect_uris':['https://claude.ai/api/mcp/auth_callback'],
|
|
'client_name':'Claude <test>', 'grant_types':['authorization_code','refresh_token'],
|
|
'response_types':['code'],'token_endpoint_auth_method':auth_method,'scope':'zpwiki:read offline_access'})
|
|
assert registered.status_code==201,registered.text
|
|
info=registered.json()
|
|
auth=await http.get('/authorize',params={'response_type':'code','client_id':info['client_id'],
|
|
'redirect_uri':'https://claude.ai/api/mcp/auth_callback','scope':'zpwiki:read offline_access',
|
|
'code_challenge':challenge,'code_challenge_method':'S256','state':state,'resource':resource})
|
|
assert auth.status_code==302,auth.text
|
|
if resource!='https://mcp.example.test/mcp':
|
|
return info,auth,None,verifier
|
|
location=auth.headers['location']
|
|
page=await http.get(location)
|
|
assert page.status_code==200
|
|
assert '<test>' in page.text and 'Claude <test>' not in page.text
|
|
assert '__Host-zp-oauth=' in page.headers['set-cookie']
|
|
assert 'Secure' in page.headers['set-cookie'] and 'HttpOnly' in page.headers['set-cookie']
|
|
import re
|
|
csrf=re.search(r'name="csrf" value="([^"]+)"',page.text)[1]
|
|
request_id=parse_qs(urlsplit(location).query)['request'][0]
|
|
return info,{'request':request_id,'csrf':csrf,'action':'allow'},page,verifier
|
|
|
|
|
|
async def oauth_grant(http, *, auth_method='none'):
|
|
from urllib.parse import parse_qs,urlsplit
|
|
info,form,page,verifier=await oauth_start(http,auth_method=auth_method)
|
|
granted=await http.post('/oauth/login',data={**form,'password':'p'*64},headers={'Origin':'https://mcp.example.test'})
|
|
assert granted.status_code==303,granted.text
|
|
query=parse_qs(urlsplit(granted.headers['location']).query)
|
|
assert query['state']==['state-123']
|
|
assert granted.headers['location'].startswith('https://claude.ai/api/mcp/auth_callback?')
|
|
body={'grant_type':'authorization_code','client_id':info['client_id'],'code':query['code'][0],
|
|
'redirect_uri':'https://claude.ai/api/mcp/auth_callback','code_verifier':verifier,
|
|
'resource':'https://mcp.example.test/mcp'}
|
|
if info.get('client_secret'):
|
|
body['client_secret']=info['client_secret']
|
|
response=await http.post('/token',data=body)
|
|
assert response.status_code==200,response.text
|
|
return info,response.json(),body
|
|
|
|
|
|
def test_oauth_discovery_and_full_flow_with_mcp(oauth_app):
|
|
import httpx
|
|
from mcp import ClientSession
|
|
from mcp.client.streamable_http import streamable_http_client
|
|
|
|
async def run():
|
|
async with oauth_app.app.router.lifespan_context(oauth_app.app):
|
|
async with httpx.AsyncClient(transport=httpx.ASGITransport(app=oauth_app),base_url='https://mcp.example.test') as http:
|
|
denied=await http.post('/mcp',json={})
|
|
assert denied.status_code==401
|
|
assert 'resource_metadata=' in denied.headers['www-authenticate']
|
|
prm=await http.get('/.well-known/oauth-protected-resource/mcp')
|
|
assert prm.status_code==200 and prm.json()['resource']=='https://mcp.example.test/mcp'
|
|
metadata=(await http.get('/.well-known/oauth-authorization-server')).json()
|
|
assert 'none' in metadata['token_endpoint_auth_methods_supported']
|
|
assert metadata['code_challenge_methods_supported']==['S256']
|
|
assert metadata['registration_endpoint'].endswith('/register')
|
|
info,token,body=await oauth_grant(http)
|
|
assert token['expires_in']>0 and token['refresh_token']
|
|
assert token['access_token'] not in oauth_app.provider.store.path.read_bytes().decode(errors='ignore')
|
|
assert (await http.post('/token',data=body)).json()['error']=='invalid_grant'
|
|
http.headers['Authorization']='Bearer '+token['access_token']
|
|
async with streamable_http_client('https://mcp.example.test/mcp',http_client=http) as (read,write,_):
|
|
async with ClientSession(read,write) as session:
|
|
await session.initialize()
|
|
assert len((await session.list_tools()).tools)==3
|
|
health=await session.call_tool('zpwiki_health')
|
|
assert not health.isError and health.structuredContent['ready']
|
|
result=await session.call_tool('zpwiki_rag',{'query':'Práca Evy Novej?'})
|
|
assert not result.isError and result.structuredContent['accepted_for_delivery']
|
|
asyncio.run(run())
|
|
|
|
|
|
def test_oauth_rejects_bad_login_csrf_and_redirect(oauth_app):
|
|
import httpx
|
|
async def run():
|
|
async with oauth_app.app.router.lifespan_context(oauth_app.app):
|
|
async with httpx.AsyncClient(transport=httpx.ASGITransport(app=oauth_app),base_url='https://mcp.example.test') as http:
|
|
bad=await http.post('/register',json={'redirect_uris':['https://evil.test/steal'],'token_endpoint_auth_method':'none'})
|
|
assert bad.status_code==400
|
|
info,form,page,verifier=await oauth_start(http)
|
|
assert page.headers['referrer-policy']=='no-referrer'
|
|
assert "frame-ancestors 'none'" in page.headers['content-security-policy']
|
|
assert (await http.post('/oauth/login',data={**form,'csrf':'wrong','password':'p'*64})).status_code==403
|
|
assert (await http.post('/oauth/login',data={**form,'password':'p'*64},headers={'Origin':'https://evil.test'})).status_code==403
|
|
wrong=await http.post('/oauth/login',data={**form,'password':'wrong'})
|
|
assert wrong.status_code==401 and 'Nesprávny' in wrong.text
|
|
for _ in range(4):
|
|
assert (await http.post('/oauth/login',data={**form,'password':'wrong'})).status_code==401
|
|
assert (await http.post('/oauth/login',data={**form,'password':'p'*64})).status_code==429
|
|
assert (await http.post('/register',content=b'not json')).status_code==400
|
|
assert (await http.post('/token',content=b'x'*16385)).status_code==413
|
|
asyncio.run(run())
|
|
|
|
|
|
def test_oauth_pkce_redirect_resource_and_expiry(oauth_app):
|
|
import httpx
|
|
from urllib.parse import parse_qs,urlsplit
|
|
async def run():
|
|
async with oauth_app.app.router.lifespan_context(oauth_app.app):
|
|
async with httpx.AsyncClient(transport=httpx.ASGITransport(app=oauth_app),base_url='https://mcp.example.test') as http:
|
|
info,auth,_,verifier=await oauth_start(http,resource='https://evil.test/mcp')
|
|
assert parse_qs(urlsplit(auth.headers['location']).query)['error']==['invalid_request']
|
|
info,form,page,verifier=await oauth_start(http)
|
|
approved=await http.post('/oauth/login',data={**form,'password':'p'*64})
|
|
code=parse_qs(urlsplit(approved.headers['location']).query)['code'][0]
|
|
request={'client_id':info['client_id'],'grant_type':'authorization_code','code':code,
|
|
'redirect_uri':'https://claude.ai/api/mcp/auth_callback','code_verifier':verifier}
|
|
bad=await http.post('/token',data={**request,'code_verifier':'bad-verifier'})
|
|
assert bad.json()['error']=='invalid_grant'
|
|
assert (await http.post('/token',data={**request,'redirect_uri':'https://evil.test'})).status_code==400
|
|
assert (await http.post('/token',data={**request,'resource':'https://evil.test/mcp'})).status_code==400
|
|
with oauth_app.provider.store.transaction() as conn:
|
|
conn.execute("UPDATE records SET expires=0 WHERE kind='code'")
|
|
assert (await http.post('/token',data=request)).json()['error']=='invalid_grant'
|
|
asyncio.run(run())
|
|
|
|
|
|
def test_oauth_refresh_replay_revocation_and_persistence(oauth_app):
|
|
import httpx
|
|
from packages.claude.oauth_provider import OAuthProvider
|
|
async def run():
|
|
async with oauth_app.app.router.lifespan_context(oauth_app.app):
|
|
async with httpx.AsyncClient(transport=httpx.ASGITransport(app=oauth_app),base_url='https://mcp.example.test') as http:
|
|
info,token,body=await oauth_grant(http)
|
|
saved=OAuthProvider('https://mcp.example.test/mcp','m'*64,'p'*64,oauth_app.provider.store.path)
|
|
assert await saved.load_access_token(token['access_token'])
|
|
refresh={'grant_type':'refresh_token','client_id':info['client_id'],
|
|
'refresh_token':token['refresh_token']}
|
|
response=await http.post('/token',data=refresh)
|
|
assert response.status_code==200,response.text
|
|
new=response.json()
|
|
assert new['refresh_token']!=token['refresh_token']
|
|
assert not await saved.load_access_token(token['access_token'])
|
|
assert await saved.load_access_token(new['access_token'])
|
|
assert (await http.post('/token',data=refresh)).json()['error']=='invalid_grant'
|
|
assert not await saved.load_access_token(new['access_token'])
|
|
info,token,body=await oauth_grant(http)
|
|
revoked=await http.post('/revoke',data={'client_id':info['client_id'],'token':token['refresh_token']})
|
|
assert revoked.status_code==200
|
|
assert not await saved.load_access_token(token['access_token'])
|
|
assert not await saved.load_refresh_token(await saved.get_client(info['client_id']),token['refresh_token'])
|
|
rotated=OAuthProvider('https://mcp.example.test/mcp','m'*64,'q'*64,oauth_app.provider.store.path)
|
|
assert not await rotated.get_client(info['client_id'])
|
|
asyncio.run(run())
|
|
|
|
|
|
@pytest.mark.parametrize('method',['client_secret_post','client_secret_basic'])
|
|
def test_oauth_confidential_clients(oauth_app,method):
|
|
import httpx,base64
|
|
async def run():
|
|
async with oauth_app.app.router.lifespan_context(oauth_app.app):
|
|
async with httpx.AsyncClient(transport=httpx.ASGITransport(app=oauth_app),base_url='https://mcp.example.test') as http:
|
|
if method=='client_secret_post':
|
|
info,token,_=await oauth_grant(http,auth_method=method)
|
|
else:
|
|
from urllib.parse import parse_qs,urlsplit
|
|
info,form,_,verifier=await oauth_start(http,auth_method=method)
|
|
granted=await http.post('/oauth/login',data={**form,'password':'p'*64})
|
|
code=parse_qs(urlsplit(granted.headers['location']).query)['code'][0]
|
|
body={'client_id':info['client_id'],'grant_type':'authorization_code','code':code,
|
|
'redirect_uri':'https://claude.ai/api/mcp/auth_callback','code_verifier':verifier}
|
|
basic=base64.b64encode((info['client_id']+':'+info['client_secret']).encode()).decode()
|
|
response=await http.post('/token',data=body,headers={'Authorization':'Basic '+basic})
|
|
assert response.status_code==200,response.text
|
|
token=response.json()
|
|
assert token['access_token']
|
|
asyncio.run(run())
|
|
|
|
|
|
def test_oauth_client_binding_scope_and_access_expiry(oauth_app):
|
|
import httpx
|
|
async def run():
|
|
async with oauth_app.app.router.lifespan_context(oauth_app.app):
|
|
async with httpx.AsyncClient(transport=httpx.ASGITransport(app=oauth_app),base_url='https://mcp.example.test') as http:
|
|
info,token,_=await oauth_grant(http)
|
|
other,_,_,_=await oauth_start(http)
|
|
refresh={'grant_type':'refresh_token','client_id':info['client_id'],
|
|
'refresh_token':token['refresh_token']}
|
|
wrong=await http.post('/token',data={**refresh,'client_id':other['client_id']})
|
|
assert wrong.json()['error']=='invalid_grant'
|
|
escalated=await http.post('/token',data={**refresh,'scope':'zpwiki:write'})
|
|
assert escalated.json()['error']=='invalid_scope'
|
|
assert (await http.post('/token',data={**refresh,'resource':'https://evil.test/mcp'})).status_code==400
|
|
with oauth_app.provider.store.transaction() as conn:
|
|
conn.execute("UPDATE records SET expires=0 WHERE kind='access'")
|
|
denied=await http.post('/mcp',json={},headers={'Authorization':'Bearer '+token['access_token']})
|
|
assert denied.status_code==401
|
|
renewed=await http.post('/token',data=refresh)
|
|
assert renewed.status_code==200
|
|
assert await oauth_app.provider.load_access_token(renewed.json()['access_token'])
|
|
static=await http.get('/health',headers={'Authorization':'Bearer '+'m'*64})
|
|
assert static.status_code==200
|
|
asyncio.run(run())
|
|
|
|
|
|
def test_oauth_cancel_expired_pending_and_missing_cookie(oauth_app):
|
|
import httpx
|
|
from urllib.parse import parse_qs,urlsplit
|
|
async def run():
|
|
async with oauth_app.app.router.lifespan_context(oauth_app.app):
|
|
async with httpx.AsyncClient(transport=httpx.ASGITransport(app=oauth_app),base_url='https://mcp.example.test') as http:
|
|
_,form,_,_=await oauth_start(http)
|
|
http.cookies.clear()
|
|
assert (await http.post('/oauth/login',data={**form,'password':'p'*64})).status_code==403
|
|
_,form,_,_=await oauth_start(http)
|
|
cancelled=await http.post('/oauth/login',data={**form,'action':'deny'})
|
|
assert cancelled.status_code==303
|
|
assert parse_qs(urlsplit(cancelled.headers['location']).query)['error']==['access_denied']
|
|
assert (await http.post('/oauth/login',data={**form,'password':'p'*64})).status_code==403
|
|
_,form,_,_=await oauth_start(http)
|
|
with oauth_app.provider.store.transaction() as conn:
|
|
conn.execute("UPDATE records SET expires=0 WHERE kind='pending'")
|
|
assert (await http.get('/oauth/login',params={'request':form['request']})).status_code==400
|
|
assert (await http.post('/oauth/login',data={**form,'password':'p'*64})).status_code==403
|
|
asyncio.run(run())
|