import os import unittest from types import SimpleNamespace from unittest.mock import MagicMock, patch import requests from web_sentinel.checks.osint import ( EmailBreachOutcome, OSINTCollector, evaluate_osint_footprint, ) from web_sentinel.model import ScanRequest class FakeResponse: def __init__( self, status_code=200, json_data=None, text="", headers=None, url="https://example.com", history=None, ): self.status_code = status_code self._json_data = json_data self.text = text self.headers = headers or {} self.url = url self.history = history or [] def json(self): if isinstance(self._json_data, Exception): raise self._json_data return self._json_data def raise_for_status(self): if self.status_code >= 400: raise requests.HTTPError(f"status {self.status_code}") class TestOSINTCollector(unittest.TestCase): def setUp(self) -> None: self.request = ScanRequest(domain="example.com") self.collector = OSINTCollector(self.request) def test_collect_certificate_transparency_subdomains(self): fake_json = [ {"name_value": "api.example.com\n*.old.example.com"}, {"name_value": "example.com\nportal.example.com"}, {"name_value": "INTERNAL.EXAMPLE.COM"}, ] response = FakeResponse(json_data=fake_json) self.collector.session.get = MagicMock(return_value=response) result = self.collector.collect_certificate_transparency_subdomains() self.assertIn("api.example.com", result) self.assertIn("portal.example.com", result) self.assertIn("internal.example.com", result) self.assertNotIn("example.com", result) def test_collect_certificate_transparency_subdomains_handles_error(self): self.collector.session.get = MagicMock(side_effect=requests.RequestException("boom")) result = self.collector.collect_certificate_transparency_subdomains() self.assertEqual(result, set()) def test_detect_web_technologies_parses_headers_and_body(self): history = [SimpleNamespace(url="https://example.com")] response = FakeResponse( headers={ "Server": "nginx/1.21.0", "X-Powered-By": "Express", "X-Generator": "Drupal", "CF-Ray": "12345", }, text="wp-content/app.js", url="https://example.com/home", history=history, ) def fake_get(url, **kwargs): return response self.collector.session.get = fake_get technologies = self.collector.detect_web_technologies() self.assertEqual(technologies["web_server"], "nginx/1.21.0") self.assertEqual(technologies["framework"], "Express") self.assertEqual(technologies["generator"], "Drupal") self.assertEqual(technologies["cms"], "WordPress") self.assertEqual(technologies["frontend_framework"], "React") self.assertEqual(technologies["cdn"], "Cloudflare") self.assertIn("https://example.com/home", technologies["probe_url"]) self.assertIn("https://example.com", technologies["redirect_chain"]) def test_check_email_breaches_missing_api_key(self): outcome = self.collector.check_email_breaches() self.assertEqual(outcome.exposures, []) self.assertEqual(outcome.skipped_reason, "missing_api_key") @patch.dict(os.environ, {"HIBP_API_KEY": "token"}) def test_check_email_breaches_with_exposure_and_errors(self): def fake_get(url, **kwargs): email = url.rsplit("/", 1)[-1] if email == "admin@example.com": return FakeResponse( json_data=[ { "Name": "BreachOne", "Title": "Breach One", "AddedDate": "2020-01-01", "PwnCount": 1000, } ] ) if email == "info@example.com": return FakeResponse(status_code=404) return FakeResponse(status_code=500) self.collector.session.get = fake_get outcome = self.collector.check_email_breaches() self.assertEqual(len(outcome.exposures), 1) exposure = outcome.exposures[0] self.assertEqual(exposure["email"], "admin@example.com") self.assertEqual(exposure["breaches"][0]["name"], "BreachOne") self.assertIsNone(outcome.skipped_reason) self.assertGreater(len(outcome.errors), 0) self.assertTrue(any("HTTP 500" in err for err in outcome.errors)) class TestEvaluateOSINTFootprint(unittest.TestCase): def test_evaluate_osint_generates_expected_findings(self): mock_collector = MagicMock() mock_collector.collect_certificate_transparency_subdomains.return_value = { f"app{i}.example.com" for i in range(12) } mock_collector.detect_web_technologies.return_value = {"web_server": "nginx"} mock_collector.check_email_breaches.return_value = EmailBreachOutcome( exposures=[{"email": "admin@example.com", "breaches": []}], errors=[], skipped_reason=None, ) with patch("web_sentinel.checks.osint.OSINTCollector", return_value=mock_collector): findings = list(evaluate_osint_footprint(ScanRequest(domain="example.com"))) self.assertEqual(len(findings), 3) i18n_keys = {finding.i18n_key for finding in findings} self.assertIn("osint.subdomains.large_footprint", i18n_keys) self.assertIn("osint.technologies.detected", i18n_keys) self.assertIn("osint.email.exposed", i18n_keys) def test_evaluate_osint_handles_skipped_email_lookup(self): mock_collector = MagicMock() mock_collector.collect_certificate_transparency_subdomains.return_value = set() mock_collector.detect_web_technologies.return_value = {} mock_collector.check_email_breaches.return_value = EmailBreachOutcome( exposures=[], errors=[], skipped_reason="missing_api_key", ) with patch("web_sentinel.checks.osint.OSINTCollector", return_value=mock_collector): findings = list(evaluate_osint_footprint(ScanRequest(domain="example.com"))) self.assertEqual(len(findings), 1) finding = findings[0] self.assertEqual(finding.i18n_key, "osint.email.lookup_skipped") self.assertEqual(finding.severity, "info") if __name__ == "__main__": unittest.main()