add public contact extraction

2026-09-03 11:15:54 +02:00
parent fb89a28f2c
commit 89eb7e07e6
14 changed files with 402 additions and 7 deletions
@@ -0,0 +1,52 @@
+import json
+import os
+import threading
+import unittest
+from http.client import HTTPConnection
+from tempfile import TemporaryDirectory
+
+from app.main import create_server
+
+
+class ContactExtractionApiTests(unittest.TestCase):
+    def setUp(self):
+        self.tmp = TemporaryDirectory()
+        os.environ['BOOTSTRAP_ADMIN_EMAIL'] = 'extract-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': 'extract-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_official_html_extracts_with_provenance_suppression_and_idempotency(self):
+        _, business = self.request('POST', '/api/v1/businesses', {'name': 'Acme', 'website': 'https://acme.test'})
+        self.request('POST', '/api/v1/suppressions', {'kind': 'email', 'value': 'sales@acme.test'})
+        payload = {'source_url': 'https://acme.test/contact', 'html': '<a href="mailto:sales@acme.test">Sales</a><p>info [at] acme [dot] test</p>', 'idempotency_key': 'extract-1'}
+        status, result = self.request('POST', f"/api/v1/businesses/{business['id']}/contacts/extract", payload)
+        self.assertEqual(status, 201); self.assertEqual(len(result['items']), 2)
+        sales = next(x for x in result['items'] if x['value'] == 'sales@acme.test')
+        self.assertTrue(sales['suppressed']); self.assertTrue(sales['do_not_contact']); self.assertEqual(sales['provenance'], 'mailto'); self.assertEqual(sales['mx_status'], 'unknown')
+        self.assertEqual(self.request('POST', f"/api/v1/businesses/{business['id']}/contacts/extract", payload)[1]['idempotent'], True)
+        self.assertEqual(self.request('GET', '/api/v1/contact-extractions?page_size=1')[1]['limit'], 1)
+
+    def test_arbitrary_and_oversized_sources_are_rejected(self):
+        _, business = self.request('POST', '/api/v1/businesses', {'name': 'Acme', 'website': 'https://acme.test'})
+        path = f"/api/v1/businesses/{business['id']}/contacts/extract"
+        self.assertEqual(self.request('POST', path, {'source_url': 'https://evil.test', 'html': '<p>x@y.test</p>'})[1]['error'], 'source_not_approved')
+        self.assertEqual(self.request('POST', path, {'source_url': 'https://acme.test', 'html': 'x' * (512 * 1024 + 1)})[0], 413)
+
+
+if __name__ == '__main__':
+    unittest.main()
@@ -0,0 +1,42 @@
+import unittest
+from app.contact_extractor import extract_contacts
+
+
+class ContactExtractorTests(unittest.TestCase):
+    def test_extracts_public_mailto_obfuscated_phone_and_ignores_false_positives(self):
+        html = '''<html><body>
+        <a href="mailto:sales@acme.test">Email Sales</a>
+        <span>info [at] acme.test</span><span>+27 (12) 345-6789</span>
+        <a href="https://wa.me/27123456789">WhatsApp</a>
+        <script>const token = 'abc@example.com';</script>
+        <img src="https://cdn.thirdparty.test/x?email=bad@thirdparty.test">
+        <span>john@example.com</span>
+        </body></html>'''
+        result = extract_contacts(html, 'https://acme.test/contact')
+        values = {(x['kind'], x['value']) for x in result}
+        self.assertIn(('email', 'sales@acme.test'), values)
+        self.assertIn(('email', 'info@acme.test'), values)
+        self.assertIn(('phone', '+27123456789'), values)
+        self.assertIn(('whatsapp', '+27123456789'), values)
+        self.assertNotIn(('email', 'abc@example.com'), values)
+        self.assertNotIn(('email', 'bad@thirdparty.test'), values)
+        self.assertNotIn(('email', 'john@example.com'), values)
+
+    def test_excludes_credentials_adjacent_to_email_like_values(self):
+        result = extract_contacts('<p>API key: foo@acme.test password: bar@acme.test</p>', 'https://acme.test/')
+        self.assertEqual(result, [])
+
+    def test_classifies_role_named_free_mail_and_suppression(self):
+        html = '<p>support@acme.test alice@acme.test bob@gmail.com</p>'
+        result = extract_contacts(html, 'https://acme.test/', suppressions=[{'kind':'email','value':'support@acme.test'}])
+        by = {x['value']: x for x in result}
+        self.assertEqual(by['support@acme.test']['classification'], 'role')
+        self.assertTrue(by['support@acme.test']['do_not_contact'])
+        self.assertTrue(by['support@acme.test']['suppressed'])
+        self.assertEqual(by['alice@acme.test']['classification'], 'named')
+        self.assertEqual(by['bob@gmail.com']['classification'], 'free_mail')
+        self.assertEqual(by['alice@acme.test']['mx_status'], 'unknown')
+
+
+if __name__ == '__main__':
+    unittest.main()