deploy updated source integrations

apps/api/app/main.py

@@ -1311,8 +1311,24 @@ class ApiHandler(BaseHTTPRequestHandler):
             else:seen.add(key);accepted.append(b)
         return self.send_json(200,{"accepted":len(accepted),"duplicates":len(rows)-len(normalized)+len(normalized)-len(accepted)-suppressed,"suppressed":suppressed,"rows":accepted})
     def list_sources(self,db,org):
-        cols='id,organization_id,name,kind,source_code,display_name,enabled,approved,policy_json,quota_json,health_status,consecutive_failures,circuit_open,last_success_at,last_failure_at,last_error,created_at,updated_at'
-        return self.send_json(200,{"organization_id":org,"items":[row_json(r) for r in db.execute(f"SELECT {cols} FROM sources WHERE organization_id=? ORDER BY id",(org,))]})
+        cols='id,organization_id,name,kind,source_code,display_name,enabled,approved,config_json,policy_json,quota_json,health_status,consecutive_failures,circuit_open,last_success_at,last_failure_at,last_error,created_at,updated_at'
+        items=[]
+        adapter_meta={a["source_code"]:a for a in available_adapters()}
+        for raw in db.execute(f"SELECT {cols} FROM sources WHERE organization_id=? ORDER BY id",(org,)):
+            item=row_json(raw); meta=adapter_meta.get(item.get("source_code") or item.get("kind"), {})
+            try: config=json.loads(raw["config_json"] or "{}")
+            except (TypeError,ValueError): config={}
+            try: policy=json.loads(raw["policy_json"] or "{}")
+            except (TypeError,ValueError): policy={}
+            item.update({"available": bool(meta.get("available", False)), "optional": bool(meta.get("optional", False)),
+                         "configured": bool(meta.get("available", False) and (config.get("csv") or config.get("rows") is not None or raw["approved"])),
+                         "credential_status": "Not required" if not meta.get("requires_credentials") else "Required / not configured",
+                         "terms_status": "Provided" if policy.get("terms_url") or config.get("terms_url") else "Not reviewed",
+                         "owner": policy.get("owner") or config.get("owner") or "Not assigned",
+                         "rate_limit": policy.get("rate_limit") or config.get("rate_limit") or "Not set"})
+            item.pop("config_json", None)
+            items.append(item)
+        return self.send_json(200,{"organization_id":org,"items":items})
     def list_queries(self,db,org):
         return self.send_json(200,{"organization_id":org,"items":[row_json(r) for r in db.execute("SELECT * FROM discovery_queries WHERE organization_id=? ORDER BY id",(org,))]})
     def list_source_records(self,db,org,q):
@@ -1342,9 +1358,23 @@ class ApiHandler(BaseHTTPRequestHandler):
         except sqlite3.IntegrityError:return self.send_json(409,{"error":"duplicate_source"})
         self.audit(db,user,'source.created',str(cur.lastrowid));db.commit();return self.send_json(201,row_json(db.execute("SELECT id,organization_id,name,kind,source_code,display_name,enabled,approved,policy_json,quota_json,health_status,consecutive_failures,circuit_open,last_success_at,last_failure_at,last_error,created_at,updated_at FROM sources WHERE id=?",(cur.lastrowid,)).fetchone()))
     def update_source(self,sid,payload,db,user):
-        if not db.execute("SELECT id FROM sources WHERE id=? AND organization_id=?",(sid,user['organization_id'])).fetchone():return self.send_json(404,{"error":"not_found"})
+        source=db.execute("SELECT * FROM sources WHERE id=? AND organization_id=?",(sid,user['organization_id'])).fetchone()
+        if not source:return self.send_json(404,{"error":"not_found"})
         if 'enabled' not in payload:return self.send_json(400,{"error":"enabled_required"})
-        value=int(bool(payload['enabled']));db.execute("UPDATE sources SET enabled=?,updated_at=CURRENT_TIMESTAMP WHERE id=?",(value,sid));self.audit(db,user,'source.enabled' if value else 'source.disabled',str(sid));db.commit();return self.send_json(200,row_json(db.execute("SELECT * FROM sources WHERE id=?",(sid,)).fetchone()))
+        value=int(bool(payload['enabled']))
+        if value:
+            adapter=adapter_for(source['kind'])
+            try: config=json.loads(source['config_json'] or '{}')
+            except (TypeError,ValueError): config={}
+            metadata=next((item for item in available_adapters() if item['source_code']==source['kind']), {})
+            if not metadata.get('available', False): return self.send_json(409,{"error":"source_unavailable"})
+            validation=adapter.validate_config(config)
+            # A blank manual source is a deliberate staging point: the query or
+            # ingest payload can provide rows later. Other adapters must be ready
+            # before they are enabled.
+            if not validation.valid and not (source['kind'] == 'manual' and not config):
+                return self.send_json(409,{"error":"source_not_configured","details":validation.errors})
+        db.execute("UPDATE sources SET enabled=?,updated_at=CURRENT_TIMESTAMP WHERE id=?",(value,sid));self.audit(db,user,'source.enabled' if value else 'source.disabled',str(sid));db.commit();return self.send_json(200,row_json(db.execute("SELECT * FROM sources WHERE id=?",(sid,)).fetchone()))
     def create_query(self,payload,db,user):
         sid=payload.get('source_id');name=str(payload.get('name','')).strip();query=payload.get('query',{})
         selected=payload.get('selected_adapters',payload.get('sources',[])); location=str(payload.get('location','')).strip(); category=str(payload.get('category','')).strip(); schedule=str(payload.get('schedule','')).strip()

apps/api/app/sources.py

@@ -7,7 +7,14 @@ Adapters never emit or persist credential values.
 from __future__ import annotations
 from dataclasses import dataclass, field
 from typing import Any, Mapping, Protocol, Sequence
-import csv, io, random, time
+import csv, io, random, time, json
+from urllib.parse import urlencode, urlparse
+from urllib.request import Request, urlopen
+
+try:
+    from .website_scanner import validate_url
+except ImportError:
+    from website_scanner import validate_url
 
 SECRET_KEYS = {"password", "passwd", "secret", "token", "api_key", "apikey", "authorization", "private_key", "credential"}
 NETWORK_KINDS = {"google_places", "bing_local", "approved_directory", "public_website", "permitted_social", "ct_logs", "dns", "rdap"}
@@ -104,6 +111,9 @@ class _Base:
     kind = ""
     source_code = ""
     display_name = ""
+    available = True
+    optional = False
+    requires_credentials = False
     def validate_config(self, config):
         if not isinstance(config, Mapping): return ValidationResult(False, ["config must be an object"])
         found = contains_secret(config)
@@ -145,9 +155,142 @@ class CsvSource(_Base):
         result = self.validate_config(config)
         if not result.valid: raise ValueError(result.errors[0])
         reader = csv.DictReader(io.StringIO(config["csv"].replace("\r\n", "\n")))
-        return DiscoveryPage([normalize_record({str(k).strip().lower():v for k,v in row.items()}) for row in reader], metadata={"adapter":self.source_code,"columns":reader.fieldnames or []})
+        records = []
+        for row in reader:
+            normalized = {str(k).strip().lower(): v for k, v in row.items()}
+            if any(str(v or '').strip() for v in normalized.values()):
+                records.append(normalize_record(normalized))
+        return DiscoveryPage(records, metadata={"adapter":self.source_code,"columns":reader.fieldnames or [],"record_count":len(records)})
+
+
+class _HttpJsonSource(_Base):
+    """Small, bounded JSON client used only for public standards-based sources."""
+    max_bytes = 256 * 1024
+    timeout = 8
+
+    def _get_json(self, url):
+        safe = validate_url(url)
+        request = Request(safe, headers={"User-Agent": "ProspectOS/0.1 (+public-source-research)"})
+        with urlopen(request, timeout=self.timeout) as response:
+            body = response.read(self.max_bytes + 1)
+            if len(body) > self.max_bytes:
+                raise ValueError("source_response_too_large")
+            return json.loads(body.decode("utf-8", "replace")), safe
+
+
+class PublicWebsiteSource(_Base):
+    kind = source_code = "public_website"
+    display_name = "Public website"
+
+    def validate_config(self, config):
+        result = super().validate_config(config)
+        if not result.valid: return result
+        urls = config.get("urls", config.get("url", []))
+        if isinstance(urls, str): urls = [urls] if urls.strip() else []
+        if not isinstance(urls, list) or not urls or len(urls) > 50:
+            return ValidationResult(False, ["urls must contain 1 to 50 public HTTP(S) URLs"])
+        for value in urls:
+            try: validate_url(str(value))
+            except ValueError: return ValidationResult(False, ["unsafe public website URL"])
+        return ValidationResult(True)
+
+    def discover(self, config, cursor=None):
+        result = self.validate_config(config)
+        if not result.valid: raise ValueError(result.errors[0])
+        urls = config.get("urls", config.get("url"))
+        if isinstance(urls, str): urls = [urls]
+        records = []
+        for raw_url in urls[:50]:
+            safe = validate_url(str(raw_url))
+            request = Request(safe, headers={"User-Agent": "ProspectOS/0.1 (+public-source-research)"})
+            with urlopen(request, timeout=8) as response:
+                body = response.read(128 * 1024).decode("utf-8", "replace")
+                final_url = response.geturl()
+            from html.parser import HTMLParser
+            parser = HTMLParser()
+            title = urlparse(final_url).hostname or safe
+            records.append(normalize_record({"name": title, "website": final_url, "description": body[:1000]}))
+        return DiscoveryPage(records, metadata={"adapter": self.source_code, "record_count": len(records), "bounded": True})
+
+
+class CtLogsSource(_HttpJsonSource):
+    kind = source_code = "ct_logs"
+    display_name = "Certificate transparency logs"
+
+    def validate_config(self, config):
+        result = super().validate_config(config)
+        if not result.valid: return result
+        query = str(config.get("domain", config.get("query", ""))).strip()
+        if not query or len(query) > 253 or any(ch in query for ch in "\r\n"):
+            return ValidationResult(False, ["domain or query is required"])
+        return ValidationResult(True)
+
+    def discover(self, config, cursor=None):
+        result = self.validate_config(config)
+        if not result.valid: raise ValueError(result.errors[0])
+        query = str(config.get("domain", config.get("query"))).strip()
+        endpoint = "https://crt.sh/?" + urlencode({"q": "%25." + query.lstrip("%.") if not query.startswith("%") else query, "output": "json"})
+        payload, source_url = self._get_json(endpoint)
+        if not isinstance(payload, list): raise ValueError("invalid_ct_response")
+        records, seen = [], set()
+        for item in payload[:500]:
+            names = str(item.get("name_value", "")) if isinstance(item, dict) else ""
+            for name in names.splitlines():
+                name = name.strip().lower().lstrip("*.")
+                if not name or name in seen or "." not in name: continue
+                seen.add(name); records.append(normalize_record({"name": name, "website": "https://" + name}))
+        return DiscoveryPage(records[:100], metadata={"adapter": self.source_code, "source_url": source_url, "record_count": len(records), "signal_only": True})
+
+
+class DnsSource(_Base):
+    kind = source_code = "dns"
+    display_name = "DNS"
+
+    def validate_config(self, config):
+        result = super().validate_config(config)
+        if not result.valid: return result
+        domains = config.get("domains", config.get("domain", []))
+        if isinstance(domains, str): domains = [domains] if domains.strip() else []
+        if not isinstance(domains, list) or not domains or len(domains) > 100: return ValidationResult(False, ["domains must contain 1 to 100 names"])
+        return ValidationResult(True)
+
+    def discover(self, config, cursor=None):
+        result = self.validate_config(config)
+        if not result.valid: raise ValueError(result.errors[0])
+        import socket
+        domains = config.get("domains", config.get("domain")); domains = [domains] if isinstance(domains, str) else domains
+        records = []
+        for domain in domains[:100]:
+            domain = str(domain).strip().lower().rstrip(".")
+            if not domain or "." not in domain: continue
+            try: addresses = sorted({item[4][0] for item in socket.getaddrinfo(domain, 443, type=socket.SOCK_STREAM)})
+            except socket.gaierror: addresses = []
+            records.append(normalize_record({"name": domain, "website": "https://" + domain, "description": json.dumps({"a_aaaa": addresses})}))
+        return DiscoveryPage(records, metadata={"adapter": self.source_code, "record_count": len(records), "dns_status_only": True})
+
+
+class RdapSource(_HttpJsonSource):
+    kind = source_code = "rdap"
+    display_name = "RDAP"
+
+    def validate_config(self, config):
+        result = super().validate_config(config)
+        if not result.valid: return result
+        domain = str(config.get("domain", "")).strip()
+        if not domain or "." not in domain: return ValidationResult(False, ["domain is required"])
+        return ValidationResult(True)
+
+    def discover(self, config, cursor=None):
+        result = self.validate_config(config)
+        if not result.valid: raise ValueError(result.errors[0])
+        domain = str(config["domain"]).strip().lower().rstrip(".")
+        payload, source_url = self._get_json("https://rdap.org/domain/" + domain)
+        return DiscoveryPage([normalize_record({"name": domain, "website": "https://" + domain, "description": json.dumps({"rdap": payload}, default=str)[:1000]})], metadata={"adapter": self.source_code, "source_url": source_url, "record_count": 1, "registration_signal_only": True})
 
 class GatedSource(_Base):
+    available = False
+    optional = True
+    requires_credentials = True
     required = "approved"
     def validate_config(self, config):
         result = super().validate_config(config)
@@ -172,11 +315,7 @@ def _gated(code, name):
 GooglePlacesSource = _gated("google_places", "Google Places")
 BingLocalSource = _gated("bing_local", "Bing / approved local API")
 ApprovedDirectorySource = _gated("approved_directory", "Approved directory")
-PublicWebsiteSource = _gated("public_website", "Public website")
 PermittedSocialSource = _gated("permitted_social", "Permitted social")
-CtLogsSource = _gated("ct_logs", "Certificate transparency logs")
-DnsSource = _gated("dns", "DNS")
-RdapSource = _gated("rdap", "RDAP")
 
 ADAPTERS = {x.source_code: x for x in (ManualSource, CsvSource, GooglePlacesSource, BingLocalSource, ApprovedDirectorySource, PublicWebsiteSource, PermittedSocialSource, CtLogsSource, DnsSource, RdapSource)}
 # common aliases used by clients
@@ -186,5 +325,9 @@ def adapter_for(kind: str) -> DiscoverySource:
     try: return ADAPTERS[str(kind).strip().lower()]()
     except KeyError: raise ValueError("unsupported source kind")
 
-def available_adapters() -> list[dict[str, str]]:
-    return [{"source_code": cls.source_code, "display_name": cls.display_name} for cls in ADAPTERS.values()]
+def available_adapters() -> list[dict[str, object]]:
+    return [{"source_code": cls.source_code, "display_name": cls.display_name,
+             "available": bool(getattr(cls, "available", False)),
+             "optional": bool(getattr(cls, "optional", False)),
+             "requires_credentials": bool(getattr(cls, "requires_credentials", False))}
+            for cls in ADAPTERS.values()]