deploy Maps v6 update
This commit is contained in:
Marco0300
@@ -0,0 +1,263 @@
|
||||
"""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 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
|
||||
TIMEOUT_SECONDS = 8
|
||||
|
||||
|
||||
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) -> 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) -> 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) -> 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": []}
|
||||
first = _fetch(url)
|
||||
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)
|
||||
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) 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
|
||||
Reference in New Issue
Block a user