diff --git a/bughunter/tools/cors_scanner.py b/bughunter/tools/cors_scanner.py index 2137c5c5..59b6d86e 100755 --- a/bughunter/tools/cors_scanner.py +++ b/bughunter/tools/cors_scanner.py @@ -131,11 +131,15 @@ def classify( """Decide whether an ACAO/ACAC response to `test` is a vuln. Pure function.""" if acao is None: return None - acac = (acac_raw or "").strip().lower() == "true" + # CORS values are case-sensitive, unlike HTTP header names. Strip only + # HTTP optional whitespace (SP/HTAB), not arbitrary Unicode whitespace. + # https://fetch.spec.whatwg.org/#cors-check + acac = (acac_raw or "").strip(" \t") == "true" + allowed_origin = acao.strip(" \t") sent = test.origin # ACAO: * — wildcard. Browsers reject *+credentials, so creds case is moot. - if acao.strip() == "*": + if allowed_origin == "*": if acac: return CorsFinding( url, sent, acao, acac, MEDIUM, @@ -148,7 +152,7 @@ def classify( "Only a finding if the endpoint returns sensitive data without a cookie.", ) - reflected = acao.strip().lower() == sent.strip().lower() + reflected = allowed_origin == sent if not reflected: return None # server returned some fixed allow-list origin, not ours diff --git a/tests/test_cors_scanner.py b/tests/test_cors_scanner.py index f4c123ae..b03a6ccf 100644 --- a/tests/test_cors_scanner.py +++ b/tests/test_cors_scanner.py @@ -1,7 +1,11 @@ """Tests for tools/cors_scanner.py — pure classification + origin generation.""" +import io +import json import os import sys +from contextlib import redirect_stdout +from unittest.mock import patch sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) @@ -13,6 +17,7 @@ CorsTest, classify, generate_tests, + main, ) URL = "https://api.target.com/me" @@ -72,6 +77,80 @@ def test_wildcard_with_credentials_flagged_as_misconfig(): assert f.severity == MEDIUM -def test_case_insensitive_reflection_match(): - f = classify(URL, _t("https://Evil.Example"), "https://evil.example", "true") - assert f is not None and f.severity == CRITICAL +def test_case_mismatched_origin_is_not_a_finding(): + # Fetch compares ACAO with the serialized Origin byte for byte. + for origin, acao in ( + ("https://evil.example", "https://Evil.Example"), + ("https://evil.example", "HTTPS://evil.example"), + ("null", "NULL"), + ("null", "Null"), + ): + assert classify(URL, _t(origin), acao, "true") is None, (origin, acao) + + +def test_credentials_value_is_case_sensitive(): + for origin in ("https://evil.example", "null"): + for acac in ("True", "TRUE", "tRuE", "\tTrue "): + f = classify(URL, _t(origin), origin, acac) + assert f is not None + assert f.acac is False, (origin, acac) + assert f.severity == MEDIUM, (origin, acac) + + +def test_wildcard_with_invalid_credentials_value_remains_info(): + f = classify(URL, _t("https://evil.example"), "*", "True") + assert f is not None + assert f.acac is False + assert f.severity == INFO + + +def test_http_optional_whitespace_is_accepted(): + for origin, severity in (("https://evil.example", CRITICAL), ("null", HIGH)): + f = classify(URL, _t(origin), f" \t{origin}\t ", " \ttrue\t ") + assert f is not None + assert f.acac is True + assert f.severity == severity + assert classify(URL, _t("https://evil.example"), " \t*\t ", None).severity == INFO + + +def test_non_http_whitespace_in_allowed_origin_is_not_ignored(): + for allowed_origin in ("https://evil.example", "null", "*"): + origin = "null" if allowed_origin == "null" else "https://evil.example" + for whitespace in ("\u00a0", "\v", "\f"): + for acao in (whitespace + allowed_origin, allowed_origin + whitespace): + assert classify(URL, _t(origin), acao, "true") is None, repr(acao) + + +def test_non_http_whitespace_does_not_enable_credentials(): + for whitespace in ("\u00a0", "\v", "\f"): + for acac in (whitespace + "true", "true" + whitespace): + f = classify(URL, _t("https://evil.example"), "https://evil.example", acac) + assert f is not None + assert f.acac is False, repr(acac) + assert f.severity == MEDIUM + + +def test_case_mismatched_responses_do_not_trigger_cli_findings(): + output = io.StringIO() + with patch( + "tools.cors_scanner._fetch_cors", + side_effect=lambda url, origin, cookie, timeout: (origin.upper(), "true"), + ) as fetch, redirect_stdout(output): + exit_code = main([URL, "--json"]) + assert fetch.call_count == len(generate_tests(URL)) + assert exit_code == 0 + assert json.loads(output.getvalue()) == [] + + +def test_valid_reflection_still_triggers_cli_findings(): + output = io.StringIO() + with patch( + "tools.cors_scanner._fetch_cors", + side_effect=lambda url, origin, cookie, timeout: (origin, "true"), + ), redirect_stdout(output): + exit_code = main([URL, "--json"]) + findings = json.loads(output.getvalue()) + assert exit_code == 2 + assert len(findings) == len(generate_tests(URL)) + assert {finding["severity"] for finding in findings} == {CRITICAL, HIGH} + assert all(finding["acac"] is True for finding in findings)