import json import unittest from unittest.mock import patch from backend import diagnostics as d class DiagnosticsTests(unittest.TestCase): def response(self, body, status=200, error=None, headers=None): return d.Response(status,json.dumps(body).encode(),headers or {},error) def test_invalid_and_private_ips(self): for value in ("127.0.0.1","::1","10.0.0.1","169.254.169.254","224.0.0.1","8.8.8.8;id","example.com","fe80::1%eth0",None): with self.subTest(value=value), self.assertRaises(ValueError): d.public_ip(value) self.assertEqual(d.public_ip(" 8.8.8.8 "),"8.8.8.8") self.assertEqual(d.public_ip("2606:4700:4700::1111"),"2606:4700:4700::1111") def test_null_scores_are_not_zero(self): for key in ("ipqs","ip2location","abuseipdb","ipapi","scamalytics","dbip"): self.assertIsNone(d.parse_provider(key,{})["score"]) self.assertEqual(d.parse_provider("ipqs",{"fraud_score":0})["score"]["value"],0) def test_three_valued_flags(self): self.assertIsNone(d.any_flag(False,None)) self.assertIs(d.any_flag(True,None),True) self.assertIs(d.any_flag(False,False),False) self.assertIsNone(d.boolean("false")) def test_ipapi_safe_numeric_parsing(self): result=d.parse_provider("ipapi",{"company":{"abuser_score":"0.001 (Low)","type":"business"},"is_vpn":False}) self.assertAlmostEqual(result["score"]["value"],0.1) self.assertIs(result["flags"]["vpn"],False) for payload in ('0);system("touch /tmp/marker");# (Low)',"nan (Low)","2.0 (Low)","-1 (Low)"): self.assertIsNone(d.parse_provider("ipapi",{"company":{"abuser_score":payload}})["score"]) def test_ipqs_uses_own_score_and_thresholds(self): for score,label in ((0,"Low"),(74,"Low"),(75,"Suspicious"),(85,"Risky"),(90,"High risk")): self.assertEqual(d.parse_provider("ipqs",{"fraud_score":score})["score"]["risk"],label) def test_dbip_category_not_fake_numeric_score(self): score=d.parse_provider("dbip",{"threatLevel":"low"})["score"] self.assertIsNone(score["value"]) self.assertEqual(score["unit"],"category") def test_missing_credentials_make_no_requests(self): def forbidden(*a,**kw): self.fail("An unconfigured provider made a network request") report=d.ip_report("8.8.8.8",env={},transport=forbidden) self.assertEqual(report["basic_status"],"not_configured") self.assertTrue(all(p["status"]=="not_configured" for p in report["providers"])) def test_provider_api_errors_and_ip_mismatch(self): env={"IPQS_KEY":"test"} cases=[(self.response({"error":"quota"}),"invalid_response"),(self.response({},429),"rate_limited"), (self.response({"ip":"1.1.1.1","fraud_score":0}),"ip_mismatch")] for response,status in cases: result=d.fetch_provider("ipqs","8.8.8.8",env,transport=lambda *a,**kw:response) self.assertEqual(result["status"],status) self.assertIsNone(result["score"]) def test_basic_exact_fields_and_target(self): data={k:"value" for k in d.BASIC_FIELDS};data.update(ip="8.8.8.8",unexpected="not exported") basic,status=d.fetch_basic("8.8.8.8",{"IPINFO_TOKEN":"test"},transport=lambda *a,**kw:self.response(data)) self.assertEqual(set(basic),set(d.BASIC_FIELDS));self.assertEqual(status,"ok") _,status=d.fetch_basic("1.1.1.1",{"IPINFO_TOKEN":"test"},transport=lambda *a,**kw:self.response(data)) self.assertEqual(status,"ip_mismatch") def test_http_success_is_not_functional_access(self): result=d.classify_service(d.SERVICES[0],self.response({"ok":True})) self.assertEqual(result["status"],"response_only") self.assertIsNone(result["region"]);self.assertIsNone(result["type"]) def test_service_failures_are_distinct(self): cases=[(self.response({},403),"access_denied"),(self.response({},429),"rate_limited"), (self.response({},401),"login_required"),(self.response({},0,error="tls_error"),"tls_error"), (self.response({},403,headers={"cf-mitigated":"challenge"}),"challenge"), (self.response({"error":{"code":"unsupported_country_region_territory"}},403),"regional_block")] for response,status in cases: self.assertEqual(d.classify_service(d.SERVICES[0],response)["status"],status) def test_generic_country_and_vpn_text_do_not_establish_block(self): response=d.Response(200,b"Read about unsupported countries or VPN policies",{}) self.assertEqual(d.classify_service(d.SERVICES[0],response)["status"],"response_only") def test_scamalytics_endpoint_rejects_foreign_host(self): for endpoint in ("http://api.scamalytics.com/account/","https://evil.invalid/","https://api.scamalytics.com.evil.invalid/"): self.assertIsNone(d.provider_endpoint("scamalytics","8.8.8.8",{"SCAMALYTICS_KEY":"test","SCAMALYTICS_ENDPOINT":endpoint})) def test_redirects_cannot_leave_exact_https_host(self): for target in ("http://chatgpt.com/", "https://chatgpt.com.evil.invalid/", "https://evil.invalid/login", "https://user@chatgpt.com/", "https://chatgpt.com:444/", "https://[bad", "/bad\npath"): with self.subTest(target=target): self.assertIsNone(d.redirect_target("https://chatgpt.com/",target,{"chatgpt.com"})) self.assertEqual(d.redirect_target("https://chatgpt.com/", "/next", {"chatgpt.com"}), "https://chatgpt.com/next") def test_api_region_does_not_override_chatgpt_web(self): calls=[] def transport(url, **kwargs): calls.append(url) return self.response({"error":{"code":"unsupported_country"}},403) if "api.openai.com" in url else self.response({}) result=d.check_service(d.SERVICES[0],transport=transport) self.assertEqual(result["status"],"response_only") self.assertEqual(result["probes"][1]["status"],"regional_block") self.assertEqual(len(calls),2) def test_challenge_stops_supplementary_requests(self): calls=[] def transport(url, **kwargs): calls.append(url) return self.response({},403,headers={"cf-mitigated":"challenge"}) result=d.check_service(d.SERVICES[0],transport=transport) self.assertEqual(result["status"],"challenge") self.assertEqual(len(calls),1) def test_redirect_limit_and_family_proxy_preserved(self): calls=[] def transport(url, **kwargs): calls.append((url,kwargs)) return self.response({},302,headers={"location":"/next"+str(len(calls))}) result=d.probe_service(d.SERVICES[0],"web","Website","https://chatgpt.com/","6","socks5h://localhost:1080",transport) self.assertEqual(result["status"],"redirect") self.assertEqual(len(calls),3) self.assertTrue(all(k["family"]=="6" and k["proxy"]=="socks5h://localhost:1080" for _,k in calls)) def test_bad_external_and_login_redirects_do_not_get_fetched(self): for target,expected in (("https://[bad","redirect"),("https://evil.invalid/login","redirect"),("/login","login_required")): calls=[] def transport(url, **kwargs): calls.append(url) return self.response({},302,headers={"location":target}) result=d.probe_service(d.SERVICES[0],"web","Website","https://chatgpt.com/","4",None,transport) self.assertEqual(result["status"],expected) self.assertEqual(len(calls),1) def test_region_hints_never_establish_functional_availability(self): service=next(s for s in d.SERVICES if s["id"]=="youtube") result=d.check_service(service,transport=lambda *a,**kw:self.response({"contentRegion":"JP"})) self.assertEqual(result["region"],"JP") self.assertEqual(result["region_source"],"page_hint") self.assertEqual(result["status"],"response_only") self.assertIsNone(result["type"]) body=b'{"contentRegion":"JP"} {"contentRegion":"US"}' result=d.check_service(service,transport=lambda *a,**kw:d.Response(200,body,{})) self.assertIsNone(result["region"]) def test_netflix_title_failure_cannot_become_playback_success(self): service=next(s for s in d.SERVICES if s["id"]=="netflix") def transport(url, **kwargs): return d.Response(error="tls_error") if "/title/" in url else self.response({}) result=d.check_service(service,transport=transport) self.assertEqual(result["status"],"response_only") self.assertTrue(all(p["status"]=="tls_error" for p in result["probes"][1:])) def test_config_quote_and_no_shell_execution(self): with self.assertRaises(ValueError):d.config_quote("value\nproxy=evil") with patch.object(d.subprocess,"run") as run: run.return_value.returncode=0;run.return_value.stdout=b"200" d.request("https://api.ipapi.is/?key=secret-test-key",headers={"X-Test":"secret-header"}) args,kwargs=run.call_args self.assertNotIn("secret-test-key"," ".join(args[0])) self.assertNotIn("secret-header"," ".join(args[0])) self.assertIn(b"secret-test-key",kwargs["input"]) self.assertNotIn("shell",kwargs) self.assertNotIn("--insecure",args[0]) if __name__=="__main__":unittest.main()