From 6269cdd6f129c7d3ea90395fd72a6d34676a96ba Mon Sep 17 00:00:00 2001 From: Marco0300 Date: Thu, 3 Sep 2026 19:47:57 +0200 Subject: [PATCH] add scoped authenticated discovery and scraping --- apps/api/app/discovery.py | 129 ++++++++++++++++++++++++ apps/api/app/main.py | 94 ++++++++++++++++- apps/api/schema.sql | 13 +++ apps/api/tests/test_scoped_discovery.py | 74 ++++++++++++++ 4 files changed, 306 insertions(+), 4 deletions(-) create mode 100644 apps/api/app/discovery.py create mode 100644 apps/api/tests/test_scoped_discovery.py diff --git a/apps/api/app/discovery.py b/apps/api/app/discovery.py new file mode 100644 index 0000000..e4e8c57 --- /dev/null +++ b/apps/api/app/discovery.py @@ -0,0 +1,129 @@ +"""Bounded, public-only prospect discovery using an explicit seed allowlist. + +There is deliberately no general web search or arbitrary URL input here: callers provide +at most a small set of public seed pages. Only links found on those seeds become candidate +sites, and subsequent crawling is same-origin, bounded, and SSRF-checked by the scanner. +""" +from __future__ import annotations + +import re +from html.parser import HTMLParser +from urllib.parse import urljoin, urldefrag, urlparse + +from .contact_extractor import extract_contacts +from .website_scanner import MAX_BYTES, MAX_REDIRECTS, _fetch, validate_url + +MAX_SEEDS = 5 +MAX_PAGES = 20 +MAX_CANDIDATES = 50 +MAX_HTML = MAX_BYTES + + +class _Links(HTMLParser): + def __init__(self): + super().__init__(convert_charrefs=True) + self.links = [] + self.title = "" + self.headings = [] + self._tag = "" + self._buf = [] + + def handle_starttag(self, tag, attrs): + tag = tag.lower(); self._tag = tag + if tag in {"a", "link"}: + attrs = {str(k).lower(): str(v or "") for k, v in attrs} + if attrs.get("href"): self.links.append(attrs["href"]) + if tag in {"title", "h1", "h2", "h3"}: self._buf = [] + + def handle_data(self, data): + if self._tag in {"title", "h1", "h2", "h3"}: self._buf.append(data) + + def handle_endtag(self, tag): + tag = tag.lower() + if tag == "title" and self._buf: self.title = " ".join("".join(self._buf).split())[:500] + if tag in {"h1", "h2", "h3"} and self._buf: self.headings.append(" ".join("".join(self._buf).split())[:300]) + self._tag = "" + + +def _html(value: bytes) -> str: + return value[:MAX_HTML].decode("utf-8", "replace") + + +def _same_site(url, root): + return (urlparse(url).hostname or "").lower().rstrip(".") == (urlparse(root).hostname or "").lower().rstrip(".") + + +def _criteria_match(text, criteria): + keywords = criteria.get("keywords", criteria.get("keyword", [])) + if isinstance(keywords, str): keywords = [keywords] + if not isinstance(keywords, list) or len(keywords) > 20: raise ValueError("invalid_criteria") + haystack = text.lower() + return not keywords or all(str(k).strip().lower() in haystack for k in keywords if str(k).strip()) + + +def discover(criteria, seed_urls, *, max_pages=MAX_PAGES, max_candidates=MAX_CANDIDATES): + if not isinstance(criteria, dict) or len(criteria) > 20: raise ValueError("invalid_criteria") + if not isinstance(seed_urls, list) or not 0 < len(seed_urls) <= MAX_SEEDS: raise ValueError("seed_urls_required") + try: max_pages = int(max_pages); max_candidates = int(max_candidates) + except (TypeError, ValueError): raise ValueError("invalid_limits") + if not 1 <= max_pages <= MAX_PAGES or not 1 <= max_candidates <= MAX_CANDIDATES: raise ValueError("invalid_limits") + seeds = [] + for raw in seed_urls: + try: safe = validate_url(raw) + except ValueError as exc: raise ValueError("unsafe_seed_url") from exc + if safe not in seeds: seeds.append(safe) + candidates = [] + for seed in seeds: + fetched = _fetch(seed, max_bytes=MAX_HTML, max_redirects=MAX_REDIRECTS) + if fetched.get("content_type") not in {"text/html", "application/xhtml+xml"}: continue + parser = _Links(); parser.feed(_html(fetched["body"])) + seed_text = " ".join([parser.title, *parser.headings]) + linked = [] + for link in parser.links: + target = urldefrag(urljoin(seed, link))[0] + if not target or target.startswith(("mailto:", "tel:", "javascript:")): continue + try: target = validate_url(target) + except ValueError: continue + if target not in linked and target not in seeds: linked.append(target) + if len(linked) >= max_candidates: break + # A seed acts as a directory/index when it links outward. If it has no + # usable links, it may itself be the explicitly allowlisted business site. + if linked: + candidates.extend(x for x in linked if x not in candidates) + elif _criteria_match(seed_text, criteria): + candidates.append(seed) + results = [] + seen = set() + for candidate in candidates[:max_candidates]: + root = candidate; queue = [candidate]; pages = [] + while queue and len(pages) < max_pages: + page = queue.pop(0) + if page in {x["url"] for x in pages}: continue + try: fetched = _fetch(page, max_bytes=MAX_HTML, max_redirects=MAX_REDIRECTS) + except ValueError: continue + if fetched.get("content_type") not in {"text/html", "application/xhtml+xml"}: continue + html = _html(fetched["body"]); parser = _Links(); parser.feed(html) + pages.append({"url": page, "final_url": fetched.get("final_url") or page, "html": html, "title": parser.title, "headings": parser.headings, "status": fetched.get("status")}) + for link in parser.links: + target = urldefrag(urljoin(page, link))[0] + if target and _same_site(target, root): + try: target = validate_url(target) + except ValueError: continue + if target not in {x["url"] for x in pages} and target not in queue: queue.append(target) + if len(queue) + len(pages) >= max_pages: queue = queue[:max(0, max_pages - len(pages))] + if not pages: continue + text = " ".join(x["title"] + " " + " ".join(x["headings"]) for x in pages) + if not _criteria_match(text, criteria): continue + domain = (urlparse(root).hostname or "").lower().removeprefix("www.") + if domain in seen: continue + seen.add(domain) + contacts = [] + evidence = [] + for page in pages: + contacts.extend(extract_contacts(page["html"], page["url"], max_results=100)) + claim = page["title"] or (page["headings"][0] if page["headings"] else "Public business page") + evidence.append({"kind": "discovery_page", "url": page["url"], "claim": claim, "provenance": "scoped_discovery"}) + deduped = {(x["kind"], x["value"]): x for x in contacts} + name = next((x["headings"][0] for x in pages if x["headings"]), next((x["title"] for x in pages if x["title"]), domain)) + results.append({"name": name[:200], "website": root, "website_domain": domain, "description": text[:1000], "contacts": list(deduped.values())[:100], "evidence": evidence, "pages": pages, "pages_crawled": len(pages), "provenance": {"mechanism": "explicit_seed_allowlist", "seed_urls": seeds, "root_url": root}}) + return {"candidates": results, "seeds": seeds, "pages_limit": max_pages, "candidate_limit": max_candidates} diff --git a/apps/api/app/main.py b/apps/api/app/main.py index 9bcc98d..ee6f38b 100644 --- a/apps/api/app/main.py +++ b/apps/api/app/main.py @@ -14,6 +14,7 @@ if __package__ in (None, ""): from app.contact_extractor import extract_contacts, MAX_HTML_BYTES, MAX_RESULTS from app.scoring import DEFAULT_RULES, signals_for_business, evaluate_score, SCORE_VERSION from app.ai_assistance import generate as generate_ai, MAX_INPUT_ITEMS, MAX_OUTPUT_CHARS + from app.discovery import discover as scoped_discover from app.config import load_config else: from .domain import deduplication_key, deduplicate_businesses, is_suppressed, normalize_business, score_business, normalize_domain, normalize_phone, match_businesses @@ -23,13 +24,14 @@ else: from .contact_extractor import extract_contacts, MAX_HTML_BYTES, MAX_RESULTS from .scoring import DEFAULT_RULES, signals_for_business, evaluate_score, SCORE_VERSION from .ai_assistance import generate as generate_ai, MAX_INPUT_ITEMS, MAX_OUTPUT_CHARS + from .discovery import discover as scoped_discover from .config import load_config ORGANIZATION_ID = "demo-tenant" SCHEMA = Path(__file__).resolve().parents[1] / "schema.sql" SESSION_DAYS = 7 PBKDF2_ITERATIONS = 300_000 MUTATING_ROLES = {"owner", "admin", "researcher"} -JOB_TYPES = {"noop", "prospect_recalculate", "source_discovery", "domain_check"} +JOB_TYPES = {"noop", "prospect_recalculate", "source_discovery", "domain_check", "scoped_discovery"} JOB_PAGE_SIZE = 100 WEBSITE_SCAN_PAGE_SIZE = 100 WEBSITE_SCAN_CACHE_SECONDS = 3600 @@ -121,8 +123,8 @@ class ApiHandler(BaseHTTPRequestHandler): db.execute("INSERT INTO audit_log (organization_id,user_id,action,details) VALUES (?,?,?,?)",(user["organization_id"],user["id"],action,details)) def business(self, db, ident, org): return db.execute("SELECT * FROM businesses WHERE id=? AND organization_id=?",(ident,org)).fetchone() def nested(self, db, bid, org): - result={"contacts":[],"domains":[],"websites":[],"evidence":[],"pipeline":[],"interactions":[],"notes":[]} - tables={"contacts":"contacts","domains":"domains","websites":"websites","evidence":"evidence","pipeline":"pipeline_entries","interactions":"interactions","notes":"notes"} + result={"contacts":[],"contact_extractions":[],"domains":[],"websites":[],"evidence":[],"pipeline":[],"interactions":[],"notes":[]} + tables={"contacts":"contacts","contact_extractions":"contact_extractions","domains":"domains","websites":"websites","evidence":"evidence","pipeline":"pipeline_entries","interactions":"interactions","notes":"notes"} for key, table in tables.items(): result[key]=[row_json(r) for r in db.execute(f"SELECT * FROM {table} WHERE business_id=? AND organization_id=? ORDER BY id",(bid,org))] return result @@ -552,6 +554,7 @@ class ApiHandler(BaseHTTPRequestHandler): if path=="/api/v1/merge-history": return self.list_merge_history(db,org) if path=="/api/v1/sources": return self.list_sources(db,org) if path=="/api/v1/discovery-queries": return self.list_queries(db,org) + if path=="/api/v1/discovery-runs": return self.list_discovery_runs(db,org,parse_qs(parsed.query)) if path=="/api/v1/source-records": return self.list_source_records(db,org,parse_qs(parsed.query)) if path=="/api/v1/jobs": return self.list_jobs(db,org,parse_qs(parsed.query)) if path=="/api/v1/domain-checks": return self.list_domain_checks(db,org,parse_qs(parsed.query)) @@ -746,6 +749,45 @@ class ApiHandler(BaseHTTPRequestHandler): rows=db.execute("SELECT * FROM jobs WHERE organization_id=? ORDER BY id DESC LIMIT ? OFFSET ?",(org,limit+1,offset)).fetchall(); more=len(rows)>limit return self.send_json(200,{"organization_id":org,"items":[job_json(r) for r in rows[:limit]],"limit":limit,"offset":offset,"has_more":more}) + def _discovery_run_json(self, row): + item = row_json(row) + for field, default in (("criteria_json", {}), ("seed_urls_json", []), ("result_json", {})): + try: item[field[:-5]] = json.loads(item.pop(field) or json.dumps(default)) + except (TypeError, ValueError): item[field[:-5]] = default + return item + + def list_discovery_runs(self, db, org, query): + try: limit = max(1, min(int((query.get("page_size") or [50])[0]), 100)); offset = max(0, int((query.get("offset") or [0])[0])) + except (ValueError, TypeError): return self.send_json(400, {"error": "invalid_pagination"}) + rows = db.execute("SELECT * FROM discovery_runs WHERE organization_id=? ORDER BY id DESC LIMIT ? OFFSET ?", (org, limit + 1, offset)).fetchall() + return self.send_json(200, {"organization_id": org, "items": [self._discovery_run_json(r) for r in rows[:limit]], "limit": limit, "offset": offset, "has_more": len(rows) > limit}) + + def create_scoped_discovery(self, payload, db, user): + criteria = payload.get("criteria", {}); seeds = payload.get("seed_urls") + if not isinstance(criteria, dict) or not isinstance(seeds, list) or not seeds: return self.send_json(400, {"error": "seed_urls_required"}) + try: + if len(seeds) > 5 or len(json.dumps(criteria).encode()) > 8192: raise ValueError("invalid_criteria") + if not isinstance(criteria.get("keywords", criteria.get("keyword", [])), (list, str)): raise ValueError("invalid_criteria") + max_pages = int(payload.get("max_pages", 20)); max_candidates = int(payload.get("max_candidates", 50)) + if not 1 <= max_pages <= 20 or not 1 <= max_candidates <= 50: raise ValueError("invalid_limits") + except (ValueError, TypeError) as exc: return self.send_json(400, {"error": str(exc) or "invalid_criteria"}) + try: + for url in seeds: validate_url(url) + except (ValueError, TypeError): + return self.send_json(400, {"error": "unsafe_seed_url"}) + key = str(payload.get("idempotency_key", "")).strip() + if not key or len(key) > 200: return self.send_json(400, {"error": "invalid_idempotency_key"}) + job_payload = {"criteria": criteria, "seed_urls": seeds, "max_pages": max_pages, "max_candidates": max_candidates} + result = self.create_job({"type": "scoped_discovery", "payload": job_payload, "idempotency_key": key, "_accepted": True, "_defer_wakeup": True}, db, user) + # create_job has already committed; read its id from the response is not available, + # so resolve by the tenant-scoped idempotency key. + job = db.execute("SELECT * FROM jobs WHERE organization_id=? AND idempotency_key=?", (user["organization_id"], key)).fetchone() + if not db.execute("SELECT id FROM discovery_runs WHERE organization_id=? AND job_id=?", (user["organization_id"], job["id"])).fetchone(): + db.execute("INSERT INTO discovery_runs(organization_id,job_id,criteria_json,seed_urls_json) VALUES(?,?,?,?)", (user["organization_id"], job["id"], json.dumps(criteria, sort_keys=True), json.dumps(seeds))) + self.audit(db, user, "discovery.created", str(job["id"])); db.commit() + getattr(self.server, "job_wakeup", threading.Event()).set() + return result + def get_job_route(self, db, org, path, query): bits=path.split("/") if len(bits)<5 or not bits[4].isdigit(): return self.send_json(404,{"error":"not_found"}) @@ -772,7 +814,8 @@ class ApiHandler(BaseHTTPRequestHandler): safe=json.dumps(redact(data),sort_keys=True,separators=(",",":")); max_attempts=max(1,min(int(payload.get("max_attempts",3)),5)) if str(payload.get("max_attempts",3)).isdigit() else 3 try: cur=db.execute("INSERT INTO jobs(organization_id,idempotency_key,type,payload,max_attempts) VALUES(?,?,?,?,?)",(user["organization_id"],key,kind,safe,max_attempts)); jid=cur.lastrowid - self.add_job_event(db,jid,user["organization_id"],"queued","Job queued",0); self.audit(db,user,"job.created",str(jid)); db.commit(); getattr(self.server,"job_wakeup",threading.Event()).set() + self.add_job_event(db,jid,user["organization_id"],"queued","Job queued",0); self.audit(db,user,"job.created",str(jid)); db.commit() + if not payload.get("_defer_wakeup"): getattr(self.server,"job_wakeup",threading.Event()).set() return self.send_json(202 if payload.get("_accepted") else 201,job_json(db.execute("SELECT * FROM jobs WHERE id=?",(jid,)).fetchone())) except sqlite3.IntegrityError: row=db.execute("SELECT * FROM jobs WHERE organization_id=? AND idempotency_key=?",(user["organization_id"],key)).fetchone(); return self.send_json(200,job_json(row)) @@ -938,6 +981,7 @@ class ApiHandler(BaseHTTPRequestHandler): bits_outreach=path.split("/") if len(bits_outreach)==7 and bits_outreach[:4]==["","api","v1","outreach"] and bits_outreach[4]=="drafts" and bits_outreach[5].isdigit() and bits_outreach[6] in {"approve","send"}: return self.approve_outreach_draft(int(bits_outreach[5]),db,user) if bits_outreach[6]=="approve" else self.send_outreach_draft(int(bits_outreach[5]),db,user) if path=="/api/v1/sources":return self.create_source(payload,db,user) + if path=="/api/v1/discovery":return self.create_scoped_discovery(payload,db,user) if path=="/api/v1/discovery-queries":return self.create_query(payload,db,user) if path=="/api/v1/suppressions":return self.create_suppression(payload,db,user) if path=="/api/v1/suppressions/import":return self.import_suppressions(payload,db,user) @@ -1262,6 +1306,41 @@ class ApiHandler(BaseHTTPRequestHandler): def log_message(self,*_):pass +def _run_scoped_discovery(db, job, handler): + payload = json.loads(job["payload"] or "{}") + result = scoped_discover(payload.get("criteria", {}), payload.get("seed_urls", []), max_pages=payload.get("max_pages", 20), max_candidates=payload.get("max_candidates", 50)) + org = job["organization_id"]; persisted = [] + for candidate in result["candidates"]: + b = normalize_business(candidate) + suppressed = is_suppressed(b, [dict(r) for r in db.execute("SELECT kind,value FROM suppressions WHERE organization_id=? AND active=1", (org,))]) + if suppressed or not b["website_domain"]: continue + existing = db.execute("SELECT id FROM businesses WHERE organization_id=? AND website_domain=?", (org, b["website_domain"])).fetchone() + if existing: bid = existing["id"] + else: + scored = score_business(b) + cur = db.execute("INSERT INTO businesses(organization_id,name,website,website_domain,email,phone,description,province,city,suburb,score,score_version,score_factors,website_class) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?)", (org,b["name"],b["website"],b["website_domain"],b["email"],b["phone"],b.get("description", ""),b["province"],b["city"],b["suburb"],scored["score"],scored["score_version"],json.dumps(scored["factors"]),scored["website_class"])) + bid = cur.lastrowid + priority = "high" if scored["score"] >= 70 else "medium" if scored["score"] >= 40 else "low" + db.execute("INSERT INTO score_history(organization_id,business_id,score,eligible,priority_band,score_version,explanations_json,signals_json) VALUES(?,?,?,?,?,?,?,?)", (org,bid,scored["score"],1,priority,scored["score_version"],json.dumps(scored["factors"]),json.dumps({"source": "scoped_discovery"}, sort_keys=True))) + scan_ids = {} + for page in candidate.get("pages", []): + scan_key = hashlib.sha256((org + ":" + page["url"]).encode()).hexdigest() + scan = db.execute("SELECT id FROM website_scans WHERE organization_id=? AND business_id=? AND cache_key=? ORDER BY id DESC LIMIT 1", (org, bid, scan_key)).fetchone() + if scan: scan_ids[page["url"]] = scan["id"]; continue + scan_result = {"input_url": page["url"], "final_url": page.get("final_url"), "status": page.get("status"), "title": page.get("title", ""), "headings": page.get("headings", []), "html": page.get("html", ""), "provenance": "scoped_discovery"} + cur_scan = db.execute("INSERT INTO website_scans(organization_id,business_id,website_id,input_url,classification,result_json,cache_key,scanned_at,cache_expires_at) VALUES(?,?,?,?,?,?,?,?,?)", (org,bid,None,page["url"],"healthy",json.dumps(scan_result, sort_keys=True),scan_key,datetime.now(timezone.utc).replace(microsecond=0).isoformat(),None)) + scan_ids[page["url"]] = cur_scan.lastrowid + for page in candidate["evidence"]: + db.execute("INSERT INTO evidence(business_id,organization_id,kind,url,claim) VALUES(?,?,?,?,?)", (bid, org, page["kind"], page["url"], page["claim"])) + for contact in candidate["contacts"]: + key = hashlib.sha256((str(job["id"]) + contact["source_url"] + contact["kind"] + contact["value"]).encode()).hexdigest() + db.execute("INSERT OR IGNORE INTO contact_extractions(organization_id,business_id,website_scan_id,extraction_key,kind,value,label,classification,confidence,source_url,public_business,mx_status,suppressed,do_not_contact,provenance) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)", (org,bid,scan_ids.get(contact["source_url"]),key,contact["kind"],contact["value"],contact["label"],contact["classification"],contact["confidence"],contact["source_url"],1,"unknown",int(contact["suppressed"]),int(contact["do_not_contact"]),contact["provenance"])) + persisted.append({"business_id": bid, "domain": b["website_domain"], "provenance": candidate["provenance"]}) + safe_result = dict(result); safe_result["candidates"] = persisted + db.execute("UPDATE discovery_runs SET result_json=?,result_count=?,updated_at=CURRENT_TIMESTAMP WHERE organization_id=? AND job_id=?", (json.dumps(safe_result, sort_keys=True), len(persisted), org, job["id"])) + handler.add_job_event(db, job["id"], org, "discovery.completed", f"Persisted {len(persisted)} candidates", 100) + + def _job_worker(server): while not server.job_stop.is_set(): db=connect(server.db_path) @@ -1275,6 +1354,13 @@ def _job_worker(server): server_handler.add_job_event(db,jid,org,"started","Job started",0); db.commit() try: payload=json.loads(job["payload"] or "{}") except ValueError: payload={} + if job["type"] == "scoped_discovery": + try: + _run_scoped_discovery(db, job, server_handler) + db.execute("UPDATE jobs SET status='succeeded',progress=100,completed_at=CURRENT_TIMESTAMP,updated_at=CURRENT_TIMESTAMP WHERE id=?", (jid,)); db.commit() + except Exception as exc: + db.execute("UPDATE jobs SET status='failed',error_code=?,completed_at=CURRENT_TIMESTAMP,updated_at=CURRENT_TIMESTAMP WHERE id=?", (str(exc)[:80] or "DISCOVERY_FAILED", jid)); server_handler.add_job_event(db,jid,org,"failed","Discovery failed",job["progress"],str(exc)[:80]); db.commit() + continue try: steps=1 if job["type"]=="noop" else max(1,min(int(payload.get("steps",5)),20)) except (ValueError,TypeError): steps=5 cancelled=False diff --git a/apps/api/schema.sql b/apps/api/schema.sql index 5b13f64..5002454 100644 --- a/apps/api/schema.sql +++ b/apps/api/schema.sql @@ -313,3 +313,16 @@ CREATE TABLE IF NOT EXISTS outreach_drafts ( ); CREATE INDEX IF NOT EXISTS idx_outreach_drafts_org ON outreach_drafts(organization_id,created_at DESC,id DESC); CREATE INDEX IF NOT EXISTS idx_outreach_drafts_business ON outreach_drafts(organization_id,business_id,id DESC); + +-- Built-in scoped discovery runs. Results are reviewable business records only; +-- this feature never sends outreach. +CREATE TABLE IF NOT EXISTS discovery_runs ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + organization_id TEXT NOT NULL REFERENCES organizations(id) ON DELETE CASCADE, + job_id INTEGER NOT NULL REFERENCES jobs(id) ON DELETE CASCADE, + criteria_json TEXT NOT NULL DEFAULT '{}', seed_urls_json TEXT NOT NULL DEFAULT '[]', + result_json TEXT NOT NULL DEFAULT '{}', result_count INTEGER NOT NULL DEFAULT 0, + created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP, updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP, + UNIQUE(organization_id,job_id) +); +CREATE INDEX IF NOT EXISTS idx_discovery_runs_org ON discovery_runs(organization_id,created_at DESC,id DESC); diff --git a/apps/api/tests/test_scoped_discovery.py b/apps/api/tests/test_scoped_discovery.py new file mode 100644 index 0000000..534756b --- /dev/null +++ b/apps/api/tests/test_scoped_discovery.py @@ -0,0 +1,74 @@ +import json +import os +import sqlite3 +import threading +import time +import unittest +from http.client import HTTPConnection +from tempfile import TemporaryDirectory +from unittest.mock import patch + +from app.main import create_server, hash_password + + +class ScopedDiscoveryApiTests(unittest.TestCase): + def setUp(self): + self.tmp = TemporaryDirectory() + os.environ['BOOTSTRAP_ADMIN_EMAIL'] = 'discover-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': 'discover-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_direct_criteria_job_crawls_allowlisted_site_and_persists_evidence(self): + pages = { + 'https://directory.test/': {'status': 200, 'final_url': 'https://directory.test/', 'content_type': 'text/html', 'body': b'

Acme Solar

Acme'}, + 'https://acme.test/': {'status': 200, 'final_url': 'https://acme.test/', 'content_type': 'text/html', 'body': b'Acme Solar

Acme Solar

Solar installers

Contact'}, + 'https://acme.test/contact': {'status': 200, 'final_url': 'https://acme.test/contact', 'content_type': 'text/html', 'body': b'

Contact Acme

Email

+27 11 555 0100

'}, + } + def fetch(url, **_): + value = pages[url]; return dict(value, redirect_chain=[], elapsed_ms=1, tls=url.startswith('https://'), certificate_status='valid') + with patch('app.discovery._fetch', side_effect=fetch), patch('app.discovery.validate_url', side_effect=lambda url, **_: url), patch('app.main.validate_url', side_effect=lambda url, **_: url): + status, created = self.request('POST', '/api/v1/discovery', {'criteria': {'keywords': ['solar'], 'location': 'Cape Town'}, 'seed_urls': ['https://directory.test/'], 'max_pages': 5, 'idempotency_key': 'scope-1'}) + self.assertEqual(status, 202); self.assertEqual(created['type'], 'scoped_discovery') + for _ in range(50): + _, job = self.request('GET', '/api/v1/jobs/' + str(created['id'])) + if job['status'] in ('succeeded', 'failed'): break + time.sleep(.02) + self.assertEqual(job['status'], 'succeeded') + status, runs = self.request('GET', '/api/v1/discovery-runs') + self.assertEqual(status, 200); self.assertEqual(runs['items'][0]['criteria']['location'], 'Cape Town') + self.assertEqual(runs['items'][0]['result_count'], 1) + status, businesses = self.request('GET', '/api/v1/businesses') + self.assertEqual(status, 200); self.assertEqual(businesses['items'][0]['website_domain'], 'acme.test') + detail = self.request('GET', '/api/v1/businesses/' + str(businesses['items'][0]['id']))[1] + self.assertTrue(any(x['url'] == 'https://acme.test/contact' for x in detail['evidence'])) + self.assertTrue(any(x['source_url'] == 'https://acme.test/contact' and x['provenance'] == 'mailto' for x in detail['contact_extractions'])) + self.assertFalse(detail.get('outreach_enabled', False)) + + def test_requires_bounded_seed_allowlist_and_rejects_ssrf(self): + self.assertEqual(self.request('POST', '/api/v1/discovery', {'criteria': {'keywords': ['x']}})[0], 400) + status, body = self.request('POST', '/api/v1/discovery', {'criteria': {}, 'seed_urls': ['http://127.0.0.1/'], 'idempotency_key': 'bad'}) + self.assertEqual(status, 400); self.assertEqual(body['error'], 'unsafe_seed_url') + + def test_results_are_tenant_isolated(self): + ph, salt = hash_password('other-password') + db = sqlite3.connect(self.server.db_path); db.execute("INSERT INTO organizations VALUES ('other-tenant','Other',CURRENT_TIMESTAMP)"); db.execute("INSERT INTO users (organization_id,email,password_hash,password_salt,role) VALUES (?,?,?,?,?)", ('other-tenant','other@example.test',ph,salt,'owner')); db.commit(); db.close() + self.cookie = None; self.request('POST', '/api/v1/auth/login', {'email': 'other@example.test', 'password': 'other-password'}) + self.assertEqual(self.request('GET', '/api/v1/discovery-runs')[1]['items'], []) + + +if __name__ == '__main__': unittest.main()