91 lines
3.3 KiB
Python
91 lines
3.3 KiB
Python
from __future__ import annotations
|
|
|
|
import contextlib
|
|
import threading
|
|
import unittest
|
|
from collections.abc import Iterator
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
|
|
from govoplan_calendar.backend.service import CalendarError, http_request
|
|
|
|
|
|
@contextlib.contextmanager
|
|
def running_http_server(handler: type[BaseHTTPRequestHandler]) -> Iterator[str]:
|
|
server = ThreadingHTTPServer(("127.0.0.1", 0), handler)
|
|
thread = threading.Thread(target=server.serve_forever, daemon=True)
|
|
thread.start()
|
|
try:
|
|
host, port = server.server_address
|
|
yield f"http://{host}:{port}"
|
|
finally:
|
|
server.shutdown()
|
|
server.server_close()
|
|
thread.join(timeout=2)
|
|
|
|
|
|
class CalendarHttpSecurityTests(unittest.TestCase):
|
|
def test_cross_origin_redirect_does_not_forward_authorization(self) -> None:
|
|
forwarded_authorization: list[str | None] = []
|
|
|
|
class TargetHandler(BaseHTTPRequestHandler):
|
|
def do_GET(self) -> None: # noqa: N802 - BaseHTTPRequestHandler API
|
|
forwarded_authorization.append(self.headers.get("Authorization"))
|
|
self.send_response(200)
|
|
self.end_headers()
|
|
|
|
def log_message(self, _format: str, *_args: object) -> None:
|
|
return
|
|
|
|
with running_http_server(TargetHandler) as target_url:
|
|
class RedirectHandler(BaseHTTPRequestHandler):
|
|
def do_GET(self) -> None: # noqa: N802 - BaseHTTPRequestHandler API
|
|
self.send_response(302)
|
|
self.send_header("Location", f"{target_url}/stolen")
|
|
self.end_headers()
|
|
|
|
def log_message(self, _format: str, *_args: object) -> None:
|
|
return
|
|
|
|
with running_http_server(RedirectHandler) as source_url:
|
|
with self.assertRaisesRegex(CalendarError, "HTTP 302"):
|
|
http_request(
|
|
f"{source_url}/feed",
|
|
headers={"Authorization": "Bearer top-secret"},
|
|
timeout=2,
|
|
)
|
|
|
|
self.assertEqual(forwarded_authorization, [])
|
|
|
|
def test_same_origin_redirect_remains_supported(self) -> None:
|
|
forwarded_authorization: list[str | None] = []
|
|
|
|
class RedirectHandler(BaseHTTPRequestHandler):
|
|
def do_GET(self) -> None: # noqa: N802 - BaseHTTPRequestHandler API
|
|
if self.path == "/feed":
|
|
self.send_response(302)
|
|
self.send_header("Location", "/calendar.ics")
|
|
self.end_headers()
|
|
return
|
|
forwarded_authorization.append(self.headers.get("Authorization"))
|
|
self.send_response(200)
|
|
self.end_headers()
|
|
self.wfile.write(b"calendar")
|
|
|
|
def log_message(self, _format: str, *_args: object) -> None:
|
|
return
|
|
|
|
with running_http_server(RedirectHandler) as source_url:
|
|
status, _headers, body = http_request(
|
|
f"{source_url}/feed",
|
|
headers={"Authorization": "Bearer expected"},
|
|
timeout=2,
|
|
)
|
|
|
|
self.assertEqual(status, 200)
|
|
self.assertEqual(body, "calendar")
|
|
self.assertEqual(forwarded_authorization, ["Bearer expected"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|