276 lines
12 KiB
Python
276 lines
12 KiB
Python
"""Safe, bounded website and domain intelligence for ProspectOS.
|
|
|
|
This module deliberately stays dependency-free. It performs small public HTTP
|
|
requests, refuses private/loopback destinations, follows only same-domain
|
|
redirects, and never treats a failed check as proof that a site does not exist.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import html
|
|
import ipaddress
|
|
import os
|
|
import re
|
|
import socket
|
|
import ssl
|
|
import time
|
|
from datetime import datetime, timezone
|
|
from html.parser import HTMLParser
|
|
from typing import Any
|
|
from urllib.error import HTTPError, URLError
|
|
from urllib.parse import urljoin, urlparse
|
|
from urllib.request import HTTPRedirectHandler, Request, build_opener
|
|
|
|
EMAIL_RE = re.compile(r"[A-Z0-9._%+\-]+@[A-Z0-9.\-]+\.[A-Z]{2,63}", re.I)
|
|
MAX_BODY_BYTES = 700_000
|
|
MAX_REDIRECTS = 2
|
|
MAX_PAGES = 2
|
|
try:
|
|
TIMEOUT_SECONDS = max(2, min(60, int(os.environ.get("WEBSITE_SCAN_TIMEOUT", "8"))))
|
|
except (TypeError, ValueError):
|
|
TIMEOUT_SECONDS = 8
|
|
# getaddrinfo has no per-call timeout parameter on all supported platforms.
|
|
# Set a bounded process default so a broken DNS server cannot hold a discovery
|
|
# worker forever during website analysis.
|
|
socket.setdefaulttimeout(TIMEOUT_SECONDS)
|
|
|
|
|
|
class _PageParser(HTMLParser):
|
|
def __init__(self) -> None:
|
|
super().__init__(convert_charrefs=True)
|
|
self.title_parts: list[str] = []
|
|
self.meta: dict[str, str] = {}
|
|
self.links: list[tuple[str, str]] = []
|
|
self._in_title = False
|
|
|
|
def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None:
|
|
values = {key.lower(): value or "" for key, value in attrs}
|
|
if tag.lower() == "title":
|
|
self._in_title = True
|
|
if tag.lower() == "meta":
|
|
key = (values.get("name") or values.get("property") or "").lower()
|
|
if key:
|
|
self.meta[key] = values.get("content", "")[:300]
|
|
if tag.lower() == "a" and values.get("href"):
|
|
self.links.append((values.get("href", "")[:500], ""))
|
|
|
|
def handle_endtag(self, tag: str) -> None:
|
|
if tag.lower() == "title":
|
|
self._in_title = False
|
|
|
|
def handle_data(self, data: str) -> None:
|
|
if self._in_title:
|
|
self.title_parts.append(data)
|
|
|
|
|
|
class _NoRedirect(HTTPRedirectHandler):
|
|
def redirect_request(self, req: Request, fp: Any, code: int, msg: str, headers: Any, newurl: str) -> None: # type: ignore[override]
|
|
return None
|
|
|
|
|
|
_opener = build_opener(_NoRedirect)
|
|
|
|
|
|
def _host_is_public(host: str) -> tuple[bool, list[str]]:
|
|
"""Resolve a hostname and reject local, reserved, or unroutable targets."""
|
|
try:
|
|
addresses = {item[4][0] for item in socket.getaddrinfo(host, None, type=socket.SOCK_STREAM)}
|
|
except (OSError, socket.gaierror):
|
|
return False, []
|
|
if not addresses:
|
|
return False, []
|
|
for raw in addresses:
|
|
try:
|
|
ip = ipaddress.ip_address(raw)
|
|
except ValueError:
|
|
return False, list(addresses)
|
|
if ip.is_private or ip.is_loopback or ip.is_link_local or ip.is_multicast or ip.is_reserved or ip.is_unspecified:
|
|
return False, list(addresses)
|
|
return True, list(addresses)
|
|
|
|
|
|
def _safe_url(value: str) -> bool:
|
|
parsed = urlparse(value)
|
|
if parsed.scheme not in {"http", "https"} or not parsed.hostname:
|
|
return False
|
|
if parsed.username or parsed.password:
|
|
return False
|
|
allowed, _ = _host_is_public(parsed.hostname)
|
|
return allowed
|
|
|
|
|
|
def _same_root(left: str, right: str) -> bool:
|
|
a = (urlparse(left).hostname or "").lower().rstrip(".")
|
|
b = (urlparse(right).hostname or "").lower().rstrip(".")
|
|
if a.startswith("www."):
|
|
a = a[4:]
|
|
if b.startswith("www."):
|
|
b = b[4:]
|
|
return bool(a and b and (a == b or a.endswith("." + b) or b.endswith("." + a)))
|
|
|
|
|
|
def _fetch(url: str, timeout_seconds: int = TIMEOUT_SECONDS) -> dict[str, Any]:
|
|
current = url
|
|
redirects = 0
|
|
started = time.monotonic()
|
|
last_status = 0
|
|
last_headers: Any = {}
|
|
body = b""
|
|
while redirects <= MAX_REDIRECTS:
|
|
if not _safe_url(current):
|
|
return {"error": "unsafe_or_unresolvable_host", "url": current, "latency_ms": int((time.monotonic() - started) * 1000)}
|
|
request = Request(current, headers={"User-Agent": "ProspectOS/1.0 (+public-site-check)", "Accept": "text/html,application/xhtml+xml"})
|
|
try:
|
|
with _opener.open(request, timeout=timeout_seconds) as response:
|
|
last_status = int(getattr(response, "status", 200) or 200)
|
|
last_headers = response.headers
|
|
body = response.read(MAX_BODY_BYTES)
|
|
final_url = response.geturl() or current
|
|
return {
|
|
"url": final_url,
|
|
"initial_url": url,
|
|
"status": last_status,
|
|
"headers": last_headers,
|
|
"body": body,
|
|
"latency_ms": int((time.monotonic() - started) * 1000),
|
|
"redirects": redirects,
|
|
}
|
|
except HTTPError as exc:
|
|
last_status = int(exc.code or 0)
|
|
last_headers = exc.headers
|
|
if 300 <= last_status < 400 and exc.headers.get("Location"):
|
|
target = urljoin(current, exc.headers.get("Location", ""))
|
|
if not _same_root(current, target) or not _safe_url(target):
|
|
return {"url": current, "initial_url": url, "status": last_status, "headers": last_headers, "body": b"", "latency_ms": int((time.monotonic() - started) * 1000), "redirects": redirects, "error": "cross_domain_redirect_blocked"}
|
|
current = target
|
|
redirects += 1
|
|
continue
|
|
return {"url": current, "initial_url": url, "status": last_status, "headers": last_headers, "body": b"", "latency_ms": int((time.monotonic() - started) * 1000), "redirects": redirects, "error": f"http_{last_status}"}
|
|
except (URLError, TimeoutError, OSError) as exc:
|
|
return {"url": current, "initial_url": url, "status": last_status, "headers": last_headers, "body": b"", "latency_ms": int((time.monotonic() - started) * 1000), "redirects": redirects, "error": str(getattr(exc, "reason", exc))[:180]}
|
|
return {"url": current, "initial_url": url, "status": last_status, "headers": last_headers, "body": body, "latency_ms": int((time.monotonic() - started) * 1000), "redirects": redirects, "error": "redirect_limit"}
|
|
|
|
|
|
def _tls_expiry(url: str, timeout_seconds: int = TIMEOUT_SECONDS) -> str:
|
|
parsed = urlparse(url)
|
|
if parsed.scheme != "https" or not parsed.hostname:
|
|
return ""
|
|
try:
|
|
context = ssl.create_default_context()
|
|
with socket.create_connection((parsed.hostname, parsed.port or 443), timeout=timeout_seconds) as raw:
|
|
with context.wrap_socket(raw, server_hostname=parsed.hostname) as secure:
|
|
cert = secure.getpeercert()
|
|
value = cert.get("notAfter") if cert else ""
|
|
if not value:
|
|
return ""
|
|
stamp = ssl.cert_time_to_seconds(value)
|
|
return datetime.fromtimestamp(stamp, timezone.utc).replace(microsecond=0).isoformat()
|
|
except (OSError, ssl.SSLError, ValueError):
|
|
return ""
|
|
|
|
|
|
def _emails(text: str) -> list[str]:
|
|
found: list[str] = []
|
|
for value in EMAIL_RE.findall(html.unescape(text or "")):
|
|
normalized = value.strip(".,;:()[]{}<>").lower()
|
|
if normalized not in found and not normalized.endswith((".png", ".jpg", ".jpeg", ".gif", ".svg")):
|
|
found.append(normalized)
|
|
return found[:30]
|
|
|
|
|
|
def scan_website(url: str, timeout_seconds: int | None = None) -> dict[str, Any]:
|
|
"""Return bounded website health, metadata, and public email evidence."""
|
|
if not url:
|
|
return {"website_status": "no_detected_website", "website_checked_at": datetime.now(timezone.utc).replace(microsecond=0).isoformat(), "website_emails": [], "email_evidence": []}
|
|
try:
|
|
timeout = max(2, min(60, int(timeout_seconds if timeout_seconds is not None else TIMEOUT_SECONDS)))
|
|
except (TypeError, ValueError):
|
|
timeout = TIMEOUT_SECONDS
|
|
first = _fetch(url, timeout)
|
|
status = int(first.get("status") or 0)
|
|
final_url = str(first.get("url") or url)
|
|
body = bytes(first.get("body") or b"")
|
|
decoded = body.decode("utf-8", "replace")
|
|
parser = _PageParser()
|
|
try:
|
|
parser.feed(decoded)
|
|
except Exception:
|
|
pass
|
|
title = " ".join(" ".join(parser.title_parts).split())[:300]
|
|
emails = _emails(decoded)
|
|
pages: list[dict[str, Any]] = [first]
|
|
if 200 <= status < 400 and len(pages) < MAX_PAGES:
|
|
for href, _ in parser.links:
|
|
candidate = urljoin(final_url, href)
|
|
label = (href or "").lower()
|
|
if any(token in label for token in ("contact", "about", "get-in-touch")) and _same_root(final_url, candidate) and _safe_url(candidate):
|
|
page = _fetch(candidate, timeout)
|
|
pages.append(page)
|
|
emails.extend(_emails(bytes(page.get("body") or b"").decode("utf-8", "replace")))
|
|
break
|
|
emails = list(dict.fromkeys(emails))[:30]
|
|
if first.get("error") and not status:
|
|
website_status = "unreachable"
|
|
elif status >= 500:
|
|
website_status = "broken"
|
|
elif status >= 400:
|
|
website_status = "error"
|
|
elif status >= 300 or int(first.get("redirects") or 0) > 0:
|
|
website_status = "redirect"
|
|
elif int(first.get("latency_ms") or 0) >= 3000:
|
|
website_status = "slow"
|
|
elif urlparse(final_url).scheme != "https":
|
|
website_status = "http_only"
|
|
else:
|
|
website_status = "healthy"
|
|
headers = first.get("headers") or {}
|
|
server = str(headers.get("Server", ""))[:120]
|
|
generator = parser.meta.get("generator", "")[:120]
|
|
technologies = ", ".join(dict.fromkeys(value for value in (server, generator) if value))[:250]
|
|
checked = datetime.now(timezone.utc).replace(microsecond=0).isoformat()
|
|
evidence = [{"email": value, "source_url": final_url, "found_at": checked, "method": "website_scan", "confidence": "high"} for value in emails]
|
|
return {
|
|
"website_status": website_status,
|
|
"website_http_status": status or None,
|
|
"website_final_url": final_url[:1000],
|
|
"website_latency_ms": int(first.get("latency_ms") or 0),
|
|
"website_https": 1 if urlparse(final_url).scheme == "https" else 0,
|
|
"website_title": title,
|
|
"website_technology": technologies,
|
|
"website_has_viewport": 1 if "viewport" in parser.meta else 0,
|
|
"website_tls_expires_at": _tls_expiry(final_url, timeout) if urlparse(final_url).scheme == "https" else "",
|
|
"website_checked_at": checked,
|
|
"website_error": str(first.get("error") or "")[:300],
|
|
"website_emails": emails,
|
|
"email_evidence": evidence,
|
|
"scanned_pages": len(pages),
|
|
}
|
|
|
|
|
|
def _slug(value: str) -> str:
|
|
value = re.sub(r"[^a-z0-9]+", "-", (value or "").lower()).strip("-")
|
|
return value[:45]
|
|
|
|
|
|
def domain_candidates(name: str, address: str = "", category: str = "") -> list[dict[str, str]]:
|
|
"""Suggest likely .co.za names; DNS state is evidence, not registrar availability."""
|
|
base = _slug(name)
|
|
city = _slug((address or "").split(",")[-1])
|
|
cat = _slug(category)
|
|
parts = [base, f"{base}-{cat}" if cat else "", f"{base}-{city}" if city else ""]
|
|
seen: set[str] = set()
|
|
output: list[dict[str, str]] = []
|
|
for stem in parts:
|
|
if not stem:
|
|
continue
|
|
domain = stem + ".co.za"
|
|
if domain in seen:
|
|
continue
|
|
seen.add(domain)
|
|
try:
|
|
socket.getaddrinfo(domain, None, type=socket.SOCK_STREAM)
|
|
dns_status = "resolves"
|
|
except OSError:
|
|
dns_status = "no_dns_response"
|
|
output.append({"domain": domain, "dns_status": dns_status, "availability_note": "DNS is only a hint; confirm availability with a registrar."})
|
|
return output
|