add ssrf-safe website analysis

2026-09-03 11:07:34 +02:00
parent f1efe39de4
commit fb89a28f2c
13 changed files with 454 additions and 15 deletions
@@ -0,0 +1,62 @@
+import json
+import os
+import threading
+import unittest
+from http.client import HTTPConnection
+from tempfile import TemporaryDirectory
+from unittest.mock import patch
+
+from app.main import create_server
+
+
+class WebsiteScanApiTests(unittest.TestCase):
+    def setUp(self):
+        self.tmp = TemporaryDirectory()
+        os.environ['BOOTSTRAP_ADMIN_EMAIL'] = 'scan-owner@example.test'
+        os.environ['BOOTSTRAP_ADMIN_PASSWORD'] = 'password'
+        self.server = create_server('127.0.0.1', 0, self.tmp.name + '/db.sqlite')
+        self.thread = threading.Thread(target=self.server.serve_forever, daemon=True); self.thread.start()
+        self.conn = HTTPConnection('127.0.0.1', self.server.server_port, timeout=4); self.cookie = None
+        self.request('POST', '/api/v1/auth/login', {'email': 'scan-owner@example.test', 'password': 'password'})
+
+    def tearDown(self):
+        self.server.shutdown(); self.server.server_close(); self.thread.join(2); self.tmp.cleanup()
+
+    def request(self, method, path, payload=None):
+        body = json.dumps(payload).encode() if payload is not None else None; headers = {'Content-Type': 'application/json'} if body else {}
+        if self.cookie: headers['Cookie'] = self.cookie
+        self.conn.request(method, path, body, headers); response = self.conn.getresponse(); cookie = response.getheader('Set-Cookie')
+        if cookie: self.cookie = cookie.split(';', 1)[0]
+        return response.status, json.loads(response.read() or b'{}')
+
+    def test_scan_is_cached_history_is_listed_and_audited(self):
+        _, business = self.request('POST', '/api/v1/businesses', {'name': 'Scan Co', 'website': 'https://scan.example'})
+        result = {'classification': 'healthy', 'status': 200, 'final_url': 'https://scan.example/', 'redirect_chain': []}
+        with patch('app.main.validate_url', return_value='https://scan.example/'), patch('app.main.scan_website', return_value=result) as scanner:
+            first_status, first = self.request('POST', f"/api/v1/businesses/{business['id']}/websites/scan", {})
+            second_status, second = self.request('POST', f"/api/v1/businesses/{business['id']}/websites/scan", {})
+        self.assertEqual(first_status, 201); self.assertEqual(second_status, 200); self.assertTrue(second['cache_hit']); scanner.assert_called_once()
+        status, history = self.request('GET', '/api/v1/website-scans?page_size=1')
+        self.assertEqual(status, 200); self.assertEqual(len(history['items']), 1); self.assertFalse(history['has_more'])
+        self.assertEqual(history['items'][0]['classification'], 'healthy')
+        self.assertEqual(self.request('GET', '/api/v1/website-scans?business_id=999999')[1]['items'], [])
+
+    def test_get_latest_scan_returns_scan_payload(self):
+        _, business = self.request('POST', '/api/v1/businesses', {'name': 'Read Scan Co', 'website': 'https://read.example'})
+        result = {'classification': 'unknown', 'status': 200, 'final_url': 'https://read.example/', 'redirect_chain': []}
+        with patch('app.main.validate_url', return_value='https://read.example/'), patch('app.main.scan_website', return_value=result):
+            self.assertEqual(self.request('POST', f"/api/v1/businesses/{business['id']}/websites/scan", {})[0], 201)
+        status, payload = self.request('GET', f"/api/v1/businesses/{business['id']}/websites/scan")
+        self.assertEqual(status, 200)
+        self.assertEqual(payload['classification'], 'unknown')
+        self.assertEqual(payload['business_id'], business['id'])
+
+    def test_unsafe_scan_is_rejected_without_fetching(self):
+        _, business = self.request('POST', '/api/v1/businesses', {'name': 'Private Scan', 'website': 'http://127.0.0.1/'})
+        with patch('app.main.scan_website') as scanner:
+            status, payload = self.request('POST', f"/api/v1/businesses/{business['id']}/websites/scan", {})
+        self.assertEqual(status, 400); self.assertEqual(payload['error'], 'unsafe_url'); scanner.assert_not_called()
+
+
+if __name__ == '__main__':
+    unittest.main()
@@ -0,0 +1,59 @@
+import unittest
+from unittest.mock import patch
+
+from app.website_scanner import classify_website, validate_url, scan_website
+
+
+class WebsiteScannerTests(unittest.TestCase):
+    def test_rejects_unsafe_schemes_and_addresses(self):
+        for url in ('file:///etc/passwd', 'ftp://example.com', 'http://127.0.0.1/', 'http://169.254.169.254/latest/meta-data'):
+            with self.subTest(url=url):
+                with self.assertRaises(ValueError):
+                    validate_url(url)
+
+    def test_classification_is_conservative(self):
+        self.assertEqual(classify_website(200, 'https://example.com', '<html><title>Acme</title><h1>Welcome</h1></html>'), 'healthy')
+        self.assertEqual(classify_website(404, 'https://example.com', ''), 'broken')
+        self.assertEqual(classify_website(200, 'https://example.com', '<html>under construction - coming soon</html>'), 'under_construction')
+        self.assertEqual(classify_website(200, 'https://example.com', '<html>domain for sale parking</html>'), 'parked')
+        self.assertEqual(classify_website(200, 'https://example.com', '<html>placeholder page</html>'), 'placeholder')
+        self.assertEqual(classify_website(301, 'https://example.com', ''), 'redirect_only')
+        self.assertEqual(classify_website(None, 'https://example.com', '', error='timeout'), 'blocked')
+        self.assertEqual(classify_website(None, 'https://example.com', ''), 'unknown')
+
+    def test_metadata_and_signals_are_extracted_from_fixture(self):
+        html = '''<!doctype html><html lang="en"><head><title>Acme</title>
+        <meta name="description" content="Solar experts"><meta name="viewport" content="width=device-width, initial-scale=1">
+        <meta name="generator" content="WordPress 6"><link rel="stylesheet" href="/style.css"></head>
+        <body><h1>Acme Solar</h1><h2>Contact us</h2><a href="/contact">Contact</a><a href="mailto:hi@acme.test">Email</a>
+        <a href="https://wa.me/27123456789">WhatsApp</a><a href="https://facebook.com/acme">Facebook</a><form action="/contact"></form></body></html>'''
+        with patch('app.website_scanner._fetch', return_value={
+            'status': 200, 'final_url': 'https://acme.test/', 'redirect_chain': [],
+            'body': html.encode(), 'content_type': 'text/html', 'elapsed_ms': 12, 'tls': True, 'certificate_status': 'valid'
+        }):
+            result = scan_website('https://acme.test/')
+        self.assertEqual(result['classification'], 'healthy')
+        self.assertEqual(result['title'], 'Acme')
+        self.assertEqual(result['meta_description'], 'Solar experts')
+        self.assertEqual(result['language'], 'en')
+        self.assertEqual(result['headings'], ['Acme Solar', 'Contact us'])
+        self.assertTrue(result['responsive_signal'])
+        self.assertIn('wordpress', result['cms_hints'])
+        self.assertTrue(result['contact_page_signal'])
+        self.assertTrue(result['form_signal'])
+        self.assertTrue(result['mail_signal'])
+        self.assertTrue(result['whatsapp_signal'])
+        self.assertTrue(result['social_signal'])
+        self.assertEqual(result['certificate_status'], 'valid')
+
+    def test_limits_are_reported_not_as_missing_contact_data(self):
+        with patch('app.website_scanner._fetch', side_effect=ValueError('response_too_large')):
+            result = scan_website('https://example.com')
+        self.assertEqual(result['classification'], 'blocked')
+        self.assertEqual(result['error_code'], 'response_too_large')
+        self.assertIsNone(result['contact_page_signal'])
+        self.assertIsNone(result['mail_signal'])
+
+
+if __name__ == '__main__':
+    unittest.main()