← zurück zur Toolbox
sitemap_gen_web_v1.py · Quelltext
21425 Bytes · SHA-256: d3b17996a539beb582b6c3a4ac72e1eae0db61b58c2016375ffe25ff644a20f4
#!/usr/home/jozapf/public_html/toolenv/bin/python
# -*- coding: utf-8 -*-
# TOOLBOX-TILE: {"title": "Sitemap Generator", "desc": "Erzeugt eine sitemap.xml durch einen begrenzten Crawl deiner Seite (bis zu 40 Seiten, Tiefe 2, nur dieselbe Domain, robots.txt wird beachtet). Ein Startpunkt für kleine Sites — kein vollständiger Crawler.", "icon": "🗺️", "type": "web", "example": "?url=example.com", "order": 18}
"""
Sitemap Generator Web — build a sitemap.xml from a bounded crawl.
A thin CGI (so the toolbox tile-discovery, which scans *.py, lists it). On a
supplied ?url= it does a small breadth-first crawl of the SAME host, honours
robots.txt, and renders a ready-to-save sitemap.xml plus a short summary.
Deliberately bounded — this is a starter for small sites, not a full crawler:
- same host only (no subdomains, no off-site)
- max 40 pages, depth 2, per-request timeout 5 s, overall budget 18 s
- only text/html 2xx pages; assets/binaries skipped
- robots.txt Disallow respected
Safety: the target host must resolve to a public IP (private/loopback/
link-local/reserved are refused → no SSRF into internal networks), only
http/https, and every fetched final URL must stay on the start host (guards
against redirect-based off-site/SSRF hops).
The executing path is protected by the shared toolbox guard.
"""
import html
import ipaddress
import os
import re
import socket
import sys
import time
from collections import deque
from email.utils import parsedate_to_datetime
from urllib.parse import parse_qs, urldefrag, urljoin, urlparse, urlsplit, urlunsplit
from urllib.robotparser import RobotFileParser
import requests
from bs4 import BeautifulSoup
# Bot/abuse guard (toolbox_guard.py sits next to this tool on the server).
# Fail open on import trouble: the guard is best-effort, never a reason to
# take a working tool offline.
try:
from toolbox_guard import guard as _guard
except Exception: # pragma: no cover
def _guard(_tool):
return True
# ═══════════════════════════════════════════════════════════════════════════
# KONFIGURATION — für die Wiederverwendung durch Dritte alles an einer Stelle.
# Wer das Tool übernimmt, passt NUR diesen Block an; der Code darunter bleibt
# unverändert. Umgebungsspezifisch außerdem: die Shebang-Zeile ganz oben und
# der optionale `toolbox_guard`-Import (ohne ihn läuft das Tool ungeschützt,
# aber voll funktionsfähig weiter).
# ═══════════════════════════════════════════════════════════════════════════
# ── Crawl-Grenzen (bewusst konservativ: Starter-Sitemap, kein Vollcrawler) ──
MAX_PAGES = 40 # höchstens so viele Seiten in die Sitemap aufnehmen
MAX_DEPTH = 2 # Link-Tiefe ab der eingegebenen Startseite
MAX_REDIRECTS = 4 # Weiterleitungen je Seite; jeder Hop wird host-geprüft
REQ_TIMEOUT = 5 # Sekunden Timeout je einzelnem Abruf
TIME_BUDGET = 18 # Sekunden Gesamtbudget, danach Abbruch (als "truncated")
MAX_BYTES = 2 * 1024 * 1024 # max. gelesene Bytes je Seite (Speicherschutz)
# ── Crawler-Identität (steht im User-Agent; robots.txt wird damit geprüft) ──
USER_AGENT = "SitemapGenWeb/1.0 (+https://jozapf.de)"
# ── Oberfläche / Branding ──
PRODUCT = "Sitemap Generator" # Produktname
BRAND = "toolbox.jozapf.de" # Marke/Domain in Titel & Footer
FONT_CSS = "https://assets.jozapf.de/css/fonts.css" # Web-Font-CSS (leer = nur System-Fonts)
SHOW_FOOTER = True # Toolbox-/Datenschutz-Footer zeigen
# ── Guard (nur mit toolbox_guard relevant; sonst ignoriert) ──
GUARD_ID = "sitemap-gen" # Kennung in der Guard-Statistik
# ═══════════════════════════════════════════════════════════════════════════
# Extensions that are never HTML pages — do not enqueue them.
_SKIP_EXT = re.compile(
r"\.(png|jpe?g|gif|webp|avif|svg|ico|css|js|mjs|json|xml|rss|atom|pdf|zip|"
r"gz|tgz|tar|rar|7z|mp4|webm|mp3|wav|ogg|woff2?|ttf|otf|eot|dmg|exe|apk)"
r"(\?|#|$)", re.I)
# DNS pinning (F5 — defeats DNS rebinding): resolve each host ONCE, validate
# every address, and pin host→IP for the rest of the process. requests then
# connects to that exact IP, while TLS SNI/cert verification still use the
# hostname. Safe here because each CGI request runs in its own short-lived
# process; crawl() clears the pins at the start of every run.
_orig_getaddrinfo = socket.getaddrinfo
_dns_pins = {}
def _pinned_getaddrinfo(host, *args, **kwargs):
return _orig_getaddrinfo(_dns_pins.get(host, host), *args, **kwargs)
socket.getaddrinfo = _pinned_getaddrinfo
def _ip_is_public(addr):
"""True if `addr` (string) is a routable public unicast IP. IPv4-mapped
IPv6 is unwrapped first (F7), so e.g. ::ffff:10.0.0.1 is seen as private."""
try:
ip = ipaddress.ip_address(addr)
except ValueError:
return False
if isinstance(ip, ipaddress.IPv6Address) and ip.ipv4_mapped is not None:
ip = ip.ipv4_mapped
return not (ip.is_private or ip.is_loopback or ip.is_link_local
or ip.is_reserved or ip.is_multicast or ip.is_unspecified)
def _safe_host(host):
"""True only if the host resolves and every address is public."""
try:
infos = _orig_getaddrinfo(host, None)
except Exception:
return False
return bool(infos) and all(_ip_is_public(i[4][0]) for i in infos)
def _pin_host(host):
"""Validate `host` (every address public) AND pin it to one IP for the rest
of the process, so the connection goes to the address we checked (no
rebinding). Returns False and pins nothing on failure or a non-public IP."""
if host in _dns_pins:
return True
try:
infos = _orig_getaddrinfo(host, None)
except Exception:
return False
if not infos or not all(_ip_is_public(i[4][0]) for i in infos):
return False
_dns_pins[host] = infos[0][4][0]
return True
def _canon(u):
"""Canonical de-duplication KEY (not the stored URL): lowercase scheme and
host, drop the default port, drop the fragment, and drop a single trailing
slash on non-root paths — so `/page` and `/page/`, `Example.com` and
`example.com`, `:443` and none all collapse to one entry. The URL actually
written to the sitemap stays the server's own (post-redirect) form."""
u, _ = urldefrag(u)
sp = urlsplit(u)
scheme = (sp.scheme or "").lower()
netloc = (sp.netloc or "").lower()
if scheme == "http" and netloc.endswith(":80"):
netloc = netloc[:-3]
elif scheme == "https" and netloc.endswith(":443"):
netloc = netloc[:-4]
path = sp.path or "/"
if len(path) > 1 and path.endswith("/"):
path = path.rstrip("/") or "/"
return urlunsplit((scheme, netloc, path, sp.query, ""))
def _lastmod(header_value):
if not header_value:
return ""
try:
return parsedate_to_datetime(header_value).strftime("%Y-%m-%d")
except (TypeError, ValueError):
return ""
def _read_capped(resp, cap):
"""Read at most `cap` bytes of the (content-encoding-decoded) body."""
chunks, total = [], 0
for chunk in resp.iter_content(16384):
if not chunk:
continue
chunks.append(chunk)
total += len(chunk)
if total >= cap:
break
return b"".join(chunks)[:cap]
def _http_get(sess, url):
"""One GET with MANUAL, host-validated redirects and a body cap.
Returns (final_url, content_type, text, lastmod) for a 2xx text/html page,
else None. Crucially, EVERY redirect hop's host is checked with _safe_host
*before* the request — so a redirect to an internal/metadata address is
never contacted (SSRF fix). The final URL is returned so the caller can pin
the canonical host (fixes apex→www losing the whole site)."""
current = url
for _ in range(MAX_REDIRECTS + 1):
pu = urlparse(current)
if pu.scheme not in ("http", "https") or not pu.hostname or not _pin_host(pu.hostname):
return None
try:
r = sess.get(current, timeout=REQ_TIMEOUT, allow_redirects=False, stream=True)
except Exception:
return None
try:
if r.status_code in (301, 302, 303, 307, 308):
loc = r.headers.get("Location")
if not loc:
return None
current = urljoin(current, loc)
continue
if r.status_code >= 400:
return None
ct = r.headers.get("Content-Type", "").split(";")[0].strip().lower()
if ct and ct != "text/html":
return None
cl = r.headers.get("Content-Length", "")
if cl.isdigit() and int(cl) > MAX_BYTES:
return None
raw = _read_capped(r, MAX_BYTES)
enc = "utf-8"
m = re.search(r"charset=([\w.:+-]+)", r.headers.get("Content-Type", ""), re.I)
if m:
enc = m.group(1)
try:
text = raw.decode(enc, errors="replace")
except (LookupError, TypeError):
text = raw.decode("utf-8", errors="replace")
return current, ct, text, _lastmod(r.headers.get("Last-Modified", ""))
finally:
r.close()
return None # too many redirects
def _load_robots(sess, origin):
"""robots.txt via the session (WITH timeout — RobotFileParser.read() has
none and can hang the request). Returns a parser, or None (= allow all)."""
rp = RobotFileParser()
try:
r = sess.get(origin + "/robots.txt", timeout=REQ_TIMEOUT, allow_redirects=True)
if r.ok and len(r.content) <= 512 * 1024:
rp.parse(r.text.splitlines())
else:
rp.parse([])
except Exception:
return None
return rp
def _enqueue_links(text, base_url, host, seen, queue, depth):
try:
soup = BeautifulSoup(text, "html.parser")
except Exception:
return
for a in soup.find_all("a", href=True):
href = a["href"].strip()
if not href or href.startswith(("mailto:", "tel:", "javascript:", "#")):
continue
nxt = urljoin(base_url, href)
np = urlparse(nxt)
if np.scheme not in ("http", "https") or np.hostname != host:
continue
if _SKIP_EXT.search(np.path or ""):
continue
key = _canon(nxt)
if key in seen:
continue
seen.add(key)
queue.append((nxt, depth))
def crawl(start):
_dns_pins.clear() # fresh pins per run
p = urlparse(start)
if p.scheme not in ("http", "https") or not p.hostname:
return None, "Bitte eine vollständige Web-Adresse angeben (z. B. example.com)."
if not _pin_host(p.hostname):
return None, ("Diese Adresse verweist auf ein internes oder privates Ziel "
"und wird aus Sicherheitsgründen nicht abgerufen.")
sess = requests.Session()
sess.headers["User-Agent"] = USER_AGENT
t0 = time.time()
# Fetch the start page FIRST — this resolves any apex→www / http→https
# redirect and establishes the canonical host (each hop host-validated).
first = _http_get(sess, start)
if first is None:
return None, ("Die Startseite ist nicht abrufbar — nicht erreichbar, kein HTML, "
"zu groß, oder eine Weiterleitung führt auf ein unsicheres Ziel.")
start_url, _ct, start_text, start_lm = first
fu = urlparse(start_url)
host = fu.hostname
origin = "%s://%s" % (fu.scheme, fu.netloc)
rp = _load_robots(sess, origin)
def allowed(u):
if rp is None:
return True
try:
return rp.can_fetch(USER_AGENT, u)
except Exception:
return True
# found: canonical-key -> (actual_url, lastmod). Keying on _canon dedups
# trailing-slash / host-case / default-port variants; the stored URL stays
# the server's own (post-redirect) form.
found, seen, queue = {}, set(), deque()
truncated, notes = False, []
seen.add(_canon(start_url))
if allowed(start_url):
found[_canon(start_url)] = (start_url, start_lm)
_enqueue_links(start_text, start_url, host, seen, queue, 1)
while queue:
if len(found) >= MAX_PAGES:
truncated = True
notes.append("Seitenlimit (%d) erreicht" % MAX_PAGES)
break
if time.time() - t0 > TIME_BUDGET:
truncated = True
notes.append("Zeitlimit erreicht")
break
url, depth = queue.popleft()
if not allowed(url):
continue
res = _http_get(sess, url)
if res is None:
continue
final_url, _ct2, text, lm = res
if urlparse(final_url).hostname != host: # off-host redirect within crawl
continue
key = _canon(final_url)
if key in found:
continue
found[key] = (final_url, lm)
if depth < MAX_DEPTH:
_enqueue_links(text, final_url, host, seen, queue, depth + 1)
return {"urls": sorted(found.values()), "truncated": truncated,
"notes": notes, "host": host}, None
def _xml_esc(s):
return (s.replace("&", "&").replace("<", "<")
.replace(">", ">").replace('"', """).replace("'", "'"))
def build_sitemap_xml(urls):
out = ['<?xml version="1.0" encoding="UTF-8"?>',
'<urlset xmlns="http://www.sitemaps.org/schemas/sitemap/0.9">']
for loc, lm in urls:
if lm:
out.append(" <url><loc>%s</loc><lastmod>%s</lastmod></url>" % (_xml_esc(loc), lm))
else:
out.append(" <url><loc>%s</loc></url>" % _xml_esc(loc))
out.append("</urlset>")
return "\n".join(out)
def build_page(url="", result=None, error=None, sitemap_xml=""):
def e(s):
return html.escape(s or "", quote=True)
body = ""
if error:
body += '<div class="card err"><strong>Hinweis:</strong> %s</div>' % e(error)
if result:
n = len(result["urls"])
note = ""
if result["truncated"]:
note = ('<p class="warn">Abgeschnitten: %s. Für größere Sites ist dies nur ein '
'Startpunkt.</p>' % e(", ".join(result["notes"])))
body += (
'<div class="card">'
'<h2>Ergebnis <span class="sub">%d Seite(n) auf %s</span></h2>%s'
'<p class="tip">Kopiere den Inhalt und speichere ihn als <code>sitemap.xml</code> '
'im Wurzelverzeichnis deiner Seite (also unter '
'<code>https://%s/sitemap.xml</code>). Trage die Adresse anschließend in deine '
'<code>robots.txt</code> und ggf. in der Google Search Console ein.</p>'
'<div class="toolbar"><button id="copy" type="button">XML kopieren</button>'
'<span id="copy_status" class="sub" aria-live="polite"></span></div>'
'<pre id="sm" tabindex="0">%s</pre>'
'</div>' % (n, e(result["host"]), note, e(result["host"]), e(sitemap_xml)))
font = '<link rel="stylesheet" href="%s">' % e(FONT_CSS) if FONT_CSS else ""
subtitle = ('Erzeugt eine <code>sitemap.xml</code> aus einem begrenzten Crawl deiner Seite '
'— bis zu %d Seiten, Tiefe %d, nur dieselbe Domain, <code>robots.txt</code> wird '
'beachtet. Ein Startpunkt für kleine Sites, kein vollständiger Crawler.'
% (MAX_PAGES, MAX_DEPTH))
if SHOW_FOOTER:
footer = ('Sitemap lieber automatisch und immer aktuell (statt manuell)? '
'<a href="https://jozapf.de" target="_blank" rel="noopener">jozapf.de</a> · '
'Teil der <a href="./">%s</a> · kein Tracking · '
'<a href="datenschutz.html">Datenschutz</a>' % e(BRAND))
else:
footer = 'kein Tracking · läuft serverseitig, keine Speicherung deiner Eingaben'
return (PAGE_TMPL
.replace("{{TITLE}}", e("%s – %s" % (PRODUCT, BRAND)))
.replace("{{FONT}}", font)
.replace("{{H1}}", e("%s | %s" % (BRAND, PRODUCT)))
.replace("{{SUBTITLE}}", subtitle)
.replace("{{FOOTER}}", footer)
.replace("{{URL}}", e(url))
.replace("{{BODY}}", body))
PAGE_TMPL = r"""<!DOCTYPE html>
<html lang="de">
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<title>{{TITLE}}</title>
{{FONT}}
<style>
:root{--bg:#0f172a;--bg2:#1e293b;--tx:#f1f5f9;--tx2:#94a3b8;--accent:#3b82f6;--accent2:#8b5cf6;--bd:#334155;--code:#0f172a;--warn:#f59e0b;--err:#f87171;}
*{box-sizing:border-box;margin:0;padding:0}
body{font-family:'Montserrat',-apple-system,BlinkMacSystemFont,'Segoe UI',Roboto,sans-serif;background:var(--bg);color:var(--tx);line-height:1.6;min-height:100vh;padding:2rem}
.container{max-width:1000px;margin:0 auto}
h1{font-size:2rem;font-weight:700;margin-bottom:.4rem;background:linear-gradient(135deg,var(--accent),var(--accent2));-webkit-background-clip:text;-webkit-text-fill-color:transparent;background-clip:text}
.subtitle{color:var(--tx2);margin-bottom:1.5rem}
h2{font-size:1.2rem;margin:0 0 1rem}
h2 .sub,.sub{font-size:.85rem;color:var(--tx2);font-weight:400}
.card{background:var(--bg2);border:1px solid var(--bd);border-radius:12px;padding:1.5rem;margin-bottom:1.25rem}
.card.err{border-color:var(--err);color:#fecaca}
label{display:block;font-size:.85rem;color:var(--tx2);margin:0 0 .35rem}
input[type=text]{width:100%;padding:.7rem .8rem;border-radius:8px;border:1px solid var(--bd);background:var(--bg);color:var(--tx);font-size:.95rem;font-family:inherit}
input:focus{outline:none;border-color:var(--accent)}
a{color:var(--accent);text-decoration:none} a:hover{text-decoration:underline}
a:focus-visible,button:focus-visible,#sm:focus-visible{outline:2px solid var(--accent);outline-offset:2px}
.btns{display:flex;flex-wrap:wrap;gap:.75rem;margin-top:1rem}
button{padding:.7rem 1.3rem;border-radius:8px;border:none;cursor:pointer;background:var(--accent);color:#fff;font-size:.95rem;font-weight:600;font-family:inherit}
button.ghost{background:transparent;border:1px solid var(--bd);color:var(--tx2)}
button:hover{filter:brightness(1.1)}
.toolbar{display:flex;align-items:center;gap:.75rem;margin:0 0 .75rem}
.tip{font-size:.85rem;color:var(--tx2);margin:.75rem 0}
.warn{color:var(--warn);font-size:.9rem;margin:.5rem 0}
code{font-family:'JetBrains Mono',Consolas,monospace;font-size:.85em;background:var(--code);color:#e879f9;padding:.15em .4em;border-radius:4px}
pre{white-space:pre-wrap;word-break:break-all;background:var(--bg);color:#a7f3d0;padding:1rem;border-radius:8px;border:1px solid var(--bd);font-size:.82rem;font-family:'JetBrains Mono',Consolas,monospace;max-height:420px;overflow:auto}
footer{text-align:center;margin-top:1.75rem;font-size:.85rem;color:var(--tx2)}
footer a{font-weight:500}
@media(max-width:640px){body{padding:1rem}h1{font-size:1.5rem}}
</style>
</head>
<body>
<div class="container">
<h1>{{H1}}</h1>
<p class="subtitle">{{SUBTITLE}}</p>
<div class="card">
<form method="get" action="">
<label>Web-Adresse</label>
<input type="text" name="url" value="{{URL}}" placeholder="example.com" autofocus>
<div class="btns"><button type="submit">Sitemap erzeugen</button></div>
</form>
</div>
{{BODY}}
<footer>{{FOOTER}}</footer>
</div>
<script src="sitemap_gen_web_v1.js"></script>
</body>
</html>
"""
def main():
params = parse_qs(os.environ.get("QUERY_STRING", ""))
url = params.get("url", [""])[0].strip()
if not url:
sys.stdout.write("Content-Type: text/html; charset=utf-8\n\n")
sys.stdout.write(build_page(url=""))
return
# Guard runs only on the executing path (this does outbound HTTP). On a
# blocked request the guard has already written the full response.
if not _guard(GUARD_ID):
return
if not re.match(r"^https?://", url, re.I):
url = "https://" + url
sys.stdout.write("Content-Type: text/html; charset=utf-8\n\n")
try:
result, error = crawl(url)
except Exception as exc:
sys.stdout.write(build_page(url=url, error="Fehler beim Crawl: %s" % exc))
return
if error:
sys.stdout.write(build_page(url=url, error=error))
return
if not result["urls"]:
sys.stdout.write(build_page(url=url, error=(
"Keine abrufbaren HTML-Seiten gefunden. Ist die Adresse erreichbar und liefert "
"sie HTML? (robots.txt kann den Zugriff auch sperren.)")))
return
xml = build_sitemap_xml(result["urls"])
sys.stdout.write(build_page(url=url, result=result, sitemap_xml=xml))
if __name__ == "__main__":
main()