| 1 | """Cross-origin redirects must drop credential headers (#1062).""" |
| 2 | |
| 3 | import http.server |
| 4 | import json |
| 5 | import socketserver |
| 6 | import threading |
| 7 | from urllib.request import Request |
| 8 | |
| 9 | from lib.http import _StripAuthOnCrossOriginRedirect, get |
| 10 | |
| 11 | |
| 12 | class _CaptureHandler(http.server.BaseHTTPRequestHandler): |
| 13 | captured: dict = {} |
| 14 | |
| 15 | def do_GET(self): |
| 16 | type(self).captured = {k.lower(): v for k, v in self.headers.items()} |
| 17 | body = b'{"ok":true}' |
| 18 | self.send_response(200) |
| 19 | self.send_header("Content-Type", "application/json") |
| 20 | self.send_header("Content-Length", str(len(body))) |
| 21 | self.end_headers() |
| 22 | self.wfile.write(body) |
| 23 | |
| 24 | def log_message(self, *_args): |
| 25 | pass |
| 26 | |
| 27 | |
| 28 | def _serve(handler): |
| 29 | httpd = socketserver.TCPServer(("127.0.0.1", 0), handler) |
| 30 | threading.Thread(target=httpd.serve_forever, daemon=True).start() |
| 31 | return httpd, httpd.server_address[1] |
| 32 | |
| 33 | |
| 34 | def _redirected_headers(src: str, dest: str, extra_headers: dict) -> dict: |
| 35 | handler = _StripAuthOnCrossOriginRedirect() |
| 36 | req = Request(src, headers=extra_headers) |
| 37 | new = handler.redirect_request(req, None, 302, "Found", {}, dest) |
| 38 | merged = {} |
| 39 | for store in (new.headers, getattr(new, "unredirected_hdrs", {})): |
| 40 | for name, value in store.items(): |
| 41 | merged[name.lower()] = value |
| 42 | return merged |
| 43 | |
| 44 | |
| 45 | def test_cross_origin_redirect_strips_authorization(): |
| 46 | class Victim(http.server.BaseHTTPRequestHandler): |
| 47 | def do_GET(self): |
| 48 | loc = f"http://127.0.0.1:{attacker.server_address[1]}/steal" |
| 49 | self.send_response(302) |
| 50 | self.send_header("Location", loc) |
| 51 | self.send_header("Content-Length", "0") |
| 52 | self.end_headers() |
| 53 | |
| 54 | def log_message(self, *_args): |
| 55 | pass |
| 56 | |
| 57 | attacker, _ap = _serve(_CaptureHandler) |
| 58 | victim = socketserver.TCPServer(("127.0.0.1", 0), Victim) |
| 59 | threading.Thread(target=victim.serve_forever, daemon=True).start() |
| 60 | vp = victim.server_address[1] |
| 61 | |
| 62 | _CaptureHandler.captured = {} |
| 63 | result = get( |
| 64 | f"http://localhost:{vp}/v1/search", |
| 65 | headers={ |
| 66 | "Authorization": "Bearer SENTINEL", |
| 67 | "X-Api-Key": "SENTINEL-KEY", |
| 68 | "X-Subscription-Token": "BRAVE-KEY", |
| 69 | }, |
| 70 | timeout=5, |
| 71 | ) |
| 72 | assert result == {"ok": True} |
| 73 | seen = {k.lower(): v for k, v in _CaptureHandler.captured.items()} |
| 74 | assert "authorization" not in seen |
| 75 | assert "x-api-key" not in seen |
| 76 | assert "x-subscription-token" not in seen |
| 77 | |
| 78 | attacker.shutdown() |
| 79 | victim.shutdown() |
| 80 | |
| 81 | |
| 82 | def test_same_origin_redirect_keeps_authorization(): |
| 83 | class SameOrigin(http.server.BaseHTTPRequestHandler): |
| 84 | captured = {} |
| 85 | |
| 86 | def do_GET(self): |
| 87 | if self.path == "/start": |
| 88 | self.send_response(302) |
| 89 | self.send_header("Location", f"http://127.0.0.1:{self.server.server_address[1]}/ok") |
| 90 | self.send_header("Content-Length", "0") |
| 91 | self.end_headers() |
| 92 | return |
| 93 | type(self).captured = {k.lower(): v for k, v in self.headers.items()} |
| 94 | body = json.dumps({"ok": True}).encode() |
| 95 | self.send_response(200) |
| 96 | self.send_header("Content-Type", "application/json") |
| 97 | self.send_header("Content-Length", str(len(body))) |
| 98 | self.end_headers() |
| 99 | self.wfile.write(body) |
| 100 | |
| 101 | def log_message(self, *_args): |
| 102 | pass |
| 103 | |
| 104 | httpd, port = _serve(SameOrigin) |
| 105 | SameOrigin.captured = {} |
| 106 | result = get( |
| 107 | f"http://127.0.0.1:{port}/start", |
| 108 | headers={"Authorization": "Bearer KEEP"}, |
| 109 | timeout=5, |
| 110 | ) |
| 111 | assert result == {"ok": True} |
| 112 | assert SameOrigin.captured.get("authorization") == "Bearer KEEP" |
| 113 | httpd.shutdown() |
| 114 | |
| 115 | |
| 116 | def test_https_to_http_same_host_strips_credentials(): |
| 117 | headers = _redirected_headers( |
| 118 | "https://api.example.com/v1", |
| 119 | "http://api.example.com/v1", |
| 120 | { |
| 121 | "Authorization": "Bearer SECRET", |
| 122 | "X-Api-Key": "KEY", |
| 123 | "X-Subscription-Token": "BRAVE", |
| 124 | }, |
| 125 | ) |
| 126 | assert "authorization" not in headers |
| 127 | assert "x-api-key" not in headers |
| 128 | assert "x-subscription-token" not in headers |
| 129 | |
| 130 | |
| 131 | def test_http_to_https_same_host_strips_credentials(): |
| 132 | headers = _redirected_headers( |
| 133 | "http://api.example.com/v1", |
| 134 | "https://api.example.com/v1", |
| 135 | {"Authorization": "Bearer SECRET"}, |
| 136 | ) |
| 137 | assert "authorization" not in headers |
| 138 | |
| 139 | |
| 140 | def test_implicit_https_port_keeps_credentials(): |
| 141 | headers = _redirected_headers( |
| 142 | "https://api.example.com/v1", |
| 143 | "https://api.example.com:443/v1", |
| 144 | {"Authorization": "Bearer KEEP"}, |
| 145 | ) |
| 146 | assert headers.get("authorization") == "Bearer KEEP" |
| 147 | |
| 148 | |
| 149 | def test_non_default_https_port_strips_credentials(): |
| 150 | headers = _redirected_headers( |
| 151 | "https://api.example.com/v1", |
| 152 | "https://api.example.com:8443/v1", |
| 153 | {"Authorization": "Bearer SECRET"}, |
| 154 | ) |
| 155 | assert "authorization" not in headers |
| 156 |