247 lines
10 KiB
Python
247 lines
10 KiB
Python
import asyncio
|
|
import unittest
|
|
from typing import Any, Optional
|
|
|
|
import httpx
|
|
from pwo import async_test
|
|
|
|
from kaya.core import HttpContext, KayaApp
|
|
from kaya.cors import CorsMixin
|
|
|
|
|
|
def make_app(**cors_kwargs: Any) -> KayaApp:
|
|
app = KayaApp(mixins=[CorsMixin(**cors_kwargs)])
|
|
|
|
@app.GET('/hello')
|
|
async def hello(ctx: HttpContext) -> None:
|
|
await ctx.send_str(200, 'Hello World!')
|
|
|
|
return app
|
|
|
|
|
|
async def request(app: KayaApp,
|
|
method: str,
|
|
path: str = '/hello',
|
|
headers: Optional[dict[str, str]] = None) -> httpx.Response:
|
|
transport = httpx.ASGITransport(app=app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://127.0.0.1:80") as client:
|
|
return await client.request(method, path, headers=headers)
|
|
|
|
|
|
class CorsSimpleRequestTest(unittest.TestCase):
|
|
|
|
@async_test
|
|
async def test_allowed_origin(self):
|
|
app = make_app(allow_origins=['https://example.com'])
|
|
r = await request(app, 'GET', headers={'Origin': 'https://example.com'})
|
|
self.assertEqual(200, r.status_code)
|
|
self.assertEqual('Hello World!', r.text)
|
|
self.assertEqual('https://example.com', r.headers.get('Access-Control-Allow-Origin'))
|
|
self.assertEqual('Origin', r.headers.get('Vary'))
|
|
|
|
@async_test
|
|
async def test_disallowed_origin(self):
|
|
app = make_app(allow_origins=['https://example.com'])
|
|
r = await request(app, 'GET', headers={'Origin': 'https://evil.com'})
|
|
self.assertEqual(200, r.status_code)
|
|
self.assertNotIn('Access-Control-Allow-Origin', r.headers)
|
|
|
|
@async_test
|
|
async def test_wildcard_origin(self):
|
|
app = make_app(allow_origins=['*'])
|
|
r = await request(app, 'GET', headers={'Origin': 'https://anything.example.com'})
|
|
self.assertEqual('*', r.headers.get('Access-Control-Allow-Origin'))
|
|
|
|
@async_test
|
|
async def test_wildcard_origin_with_credentials_echoes_origin(self):
|
|
app = make_app(allow_origins=['*'], allow_credentials=True)
|
|
r = await request(app, 'GET', headers={'Origin': 'https://example.com'})
|
|
self.assertEqual('https://example.com', r.headers.get('Access-Control-Allow-Origin'))
|
|
self.assertEqual('true', r.headers.get('Access-Control-Allow-Credentials'))
|
|
self.assertEqual('Origin', r.headers.get('Vary'))
|
|
|
|
@async_test
|
|
async def test_origin_regex(self):
|
|
app = make_app(allow_origin_regex=r'https://.*\.example\.com')
|
|
r = await request(app, 'GET', headers={'Origin': 'https://api.example.com'})
|
|
self.assertEqual('https://api.example.com', r.headers.get('Access-Control-Allow-Origin'))
|
|
r = await request(app, 'GET', headers={'Origin': 'https://example.com.evil.org'})
|
|
self.assertNotIn('Access-Control-Allow-Origin', r.headers)
|
|
|
|
@async_test
|
|
async def test_no_origin_header(self):
|
|
app = make_app(allow_origins=['*'])
|
|
r = await request(app, 'GET')
|
|
self.assertEqual(200, r.status_code)
|
|
self.assertNotIn('Access-Control-Allow-Origin', r.headers)
|
|
|
|
@async_test
|
|
async def test_expose_headers(self):
|
|
app = make_app(allow_origins=['*'], expose_headers=['X-Total-Count'])
|
|
r = await request(app, 'GET', headers={'Origin': 'https://example.com'})
|
|
self.assertEqual('X-Total-Count', r.headers.get('Access-Control-Expose-Headers'))
|
|
|
|
@async_test
|
|
async def test_handler_set_cors_header_not_overwritten(self):
|
|
app = KayaApp(mixins=[CorsMixin(allow_origins=['*'])])
|
|
|
|
@app.GET('/custom')
|
|
async def custom(ctx: HttpContext) -> None:
|
|
await ctx.send_str(200, 'custom', headers={'Access-Control-Allow-Origin': 'https://custom.example.com'})
|
|
|
|
r = await request(app, 'GET', '/custom', headers={'Origin': 'https://example.com'})
|
|
self.assertEqual('https://custom.example.com', r.headers.get('Access-Control-Allow-Origin'))
|
|
|
|
|
|
class CorsPreflightTest(unittest.TestCase):
|
|
|
|
@staticmethod
|
|
def preflight_headers(origin: str = 'https://example.com',
|
|
method: str = 'POST',
|
|
headers: Optional[str] = None) -> dict[str, str]:
|
|
result = {
|
|
'Origin': origin,
|
|
'Access-Control-Request-Method': method,
|
|
}
|
|
if headers is not None:
|
|
result['Access-Control-Request-Headers'] = headers
|
|
return result
|
|
|
|
@async_test
|
|
async def test_preflight_allowed(self):
|
|
app = make_app(allow_origins=['https://example.com'],
|
|
allow_methods=('GET', 'POST'),
|
|
allow_credentials=True)
|
|
r = await request(app, 'OPTIONS', headers=self.preflight_headers())
|
|
self.assertEqual(200, r.status_code)
|
|
self.assertEqual('OK', r.text)
|
|
self.assertEqual('https://example.com', r.headers.get('Access-Control-Allow-Origin'))
|
|
self.assertEqual('GET, POST', r.headers.get('Access-Control-Allow-Methods'))
|
|
self.assertEqual('600', r.headers.get('Access-Control-Max-Age'))
|
|
self.assertEqual('true', r.headers.get('Access-Control-Allow-Credentials'))
|
|
self.assertEqual('Origin', r.headers.get('Vary'))
|
|
|
|
@async_test
|
|
async def test_preflight_wildcard_origin(self):
|
|
app = make_app(allow_origins=['*'], allow_methods=('GET', 'POST'))
|
|
r = await request(app, 'OPTIONS', headers=self.preflight_headers())
|
|
self.assertEqual(200, r.status_code)
|
|
self.assertEqual('*', r.headers.get('Access-Control-Allow-Origin'))
|
|
|
|
@async_test
|
|
async def test_preflight_disallowed_origin(self):
|
|
app = make_app(allow_origins=['https://example.com'], allow_methods=('GET', 'POST'))
|
|
r = await request(app, 'OPTIONS', headers=self.preflight_headers(origin='https://evil.com'))
|
|
self.assertEqual(400, r.status_code)
|
|
self.assertEqual('Disallowed CORS origin', r.text)
|
|
|
|
@async_test
|
|
async def test_preflight_disallowed_method(self):
|
|
app = make_app(allow_origins=['https://example.com'], allow_methods=('GET',))
|
|
r = await request(app, 'OPTIONS', headers=self.preflight_headers(method='DELETE'))
|
|
self.assertEqual(400, r.status_code)
|
|
self.assertEqual('Disallowed CORS method', r.text)
|
|
|
|
@async_test
|
|
async def test_preflight_disallowed_headers(self):
|
|
app = make_app(allow_origins=['https://example.com'], allow_methods=('GET', 'POST'))
|
|
r = await request(app, 'OPTIONS', headers=self.preflight_headers(headers='X-Custom'))
|
|
self.assertEqual(400, r.status_code)
|
|
self.assertEqual('Disallowed CORS headers', r.text)
|
|
|
|
@async_test
|
|
async def test_preflight_safelisted_headers_allowed(self):
|
|
app = make_app(allow_origins=['https://example.com'], allow_methods=('GET', 'POST'))
|
|
r = await request(app, 'OPTIONS', headers=self.preflight_headers(headers='Content-Type'))
|
|
self.assertEqual(200, r.status_code)
|
|
|
|
@async_test
|
|
async def test_preflight_allow_all_headers_mirrors_request(self):
|
|
app = make_app(allow_origins=['*'], allow_methods=('GET', 'POST'), allow_headers=['*'])
|
|
r = await request(app, 'OPTIONS', headers=self.preflight_headers(headers='X-Custom, X-Other'))
|
|
self.assertEqual(200, r.status_code)
|
|
self.assertEqual('X-Custom, X-Other', r.headers.get('Access-Control-Allow-Headers'))
|
|
|
|
@async_test
|
|
async def test_preflight_configured_allow_headers(self):
|
|
app = make_app(allow_origins=['https://example.com'],
|
|
allow_methods=('GET', 'POST'),
|
|
allow_headers=('X-Custom',))
|
|
r = await request(app, 'OPTIONS', headers=self.preflight_headers(headers='X-Custom'))
|
|
self.assertEqual(200, r.status_code)
|
|
allow_headers = r.headers.get('Access-Control-Allow-Headers')
|
|
self.assertIsNotNone(allow_headers)
|
|
assert allow_headers is not None
|
|
self.assertIn('X-Custom', allow_headers)
|
|
|
|
@async_test
|
|
async def test_preflight_response_not_overwritten_by_handler(self):
|
|
# even when a user-registered OPTIONS handler matches, the preflight
|
|
# response sent by the mixin is the only one delivered to the client
|
|
app = KayaApp(mixins=[CorsMixin(allow_origins=['https://example.com'], allow_methods=('GET', 'POST'))])
|
|
|
|
@app.GET('/hello')
|
|
async def hello(ctx: HttpContext) -> None:
|
|
await ctx.send_str(200, 'Hello World!')
|
|
|
|
@app.OPTIONS('/hello')
|
|
async def options(ctx: HttpContext) -> None:
|
|
await ctx.send_str(200, 'custom OPTIONS handler')
|
|
|
|
r = await request(app, 'OPTIONS', headers=self.preflight_headers())
|
|
self.assertEqual(200, r.status_code)
|
|
self.assertEqual('OK', r.text)
|
|
|
|
@async_test
|
|
async def test_options_without_preflight_headers_routes_normally(self):
|
|
app = KayaApp(mixins=[CorsMixin(allow_origins=['https://example.com'])])
|
|
|
|
@app.OPTIONS('/hello')
|
|
async def options(ctx: HttpContext) -> None:
|
|
await ctx.send_str(200, 'custom OPTIONS handler')
|
|
|
|
# an OPTIONS request without Access-Control-Request-Method is not a
|
|
# preflight request and is routed normally
|
|
r = await request(app, 'OPTIONS', headers={'Origin': 'https://example.com'})
|
|
self.assertEqual(200, r.status_code)
|
|
self.assertEqual('custom OPTIONS handler', r.text)
|
|
self.assertEqual('https://example.com', r.headers.get('Access-Control-Allow-Origin'))
|
|
|
|
|
|
class CorsRsgiTest(unittest.TestCase):
|
|
|
|
def test_rsgi_context_header_injection(self):
|
|
from kaya.rsgi import RsgiContext
|
|
|
|
class FakeScope:
|
|
scheme = 'http'
|
|
method = 'GET'
|
|
path = '/'
|
|
query_string = ''
|
|
headers = {'origin': 'https://example.com'}
|
|
client = '127.0.0.1:12345'
|
|
server = '127.0.0.1:80'
|
|
|
|
class FakeProtocol:
|
|
def __init__(self) -> None:
|
|
self.responses = []
|
|
|
|
def response_str(self, status: int, headers: list, body: str) -> None:
|
|
self.responses.append((status, dict(headers), body))
|
|
|
|
mixin = CorsMixin(allow_origins=['https://example.com'])
|
|
protocol = FakeProtocol()
|
|
ctx = RsgiContext(FakeScope(), protocol) # type: ignore[arg-type]
|
|
|
|
async def run() -> None:
|
|
wrapped = await mixin._before_request(ctx)
|
|
assert wrapped is not None
|
|
await wrapped.send_str(200, 'hi')
|
|
|
|
asyncio.run(run())
|
|
self.assertEqual(1, len(protocol.responses))
|
|
status, headers, body = protocol.responses[0]
|
|
self.assertEqual(200, status)
|
|
self.assertEqual('https://example.com', headers.get('Access-Control-Allow-Origin'))
|
|
self.assertEqual('Origin', headers.get('Vary'))
|